You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by st...@apache.org on 2022/03/23 00:42:19 UTC
[impala] branch master updated (a5a9b1e -> 7273c9f)
This is an automated email from the ASF dual-hosted git repository.
stigahuang pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git.
from a5a9b1e IMPALA-11110: Switch debug builds to use -Og
new 07a3e6e IMPALA-10992 Planner changes for estimate peak memory
new e90cf9c IMPALA-11119: [DOCS] document BYTES function
new 7273c9f IMPALA-11192: Batch uploading files in test_scanner_fuzz.py
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
be/src/service/query-options.cc | 8 +
be/src/service/query-options.h | 6 +-
common/thrift/Frontend.thrift | 7 +
common/thrift/ImpalaService.thrift | 6 +
common/thrift/Query.thrift | 11 +
docs/topics/impala_string_functions.xml | 37 ++
.../apache/impala/analysis/AnalysisContext.java | 25 +-
.../java/org/apache/impala/analysis/Analyzer.java | 16 +
.../org/apache/impala/planner/HBaseScanNode.java | 2 +-
.../org/apache/impala/planner/HdfsScanNode.java | 14 +-
.../org/apache/impala/planner/KuduScanNode.java | 9 +-
.../impala/planner/ResourceProfileBuilder.java | 1 +
.../java/org/apache/impala/service/Frontend.java | 450 +++++++++++++++++++--
.../java/org/apache/impala/util/ClassUtil.java | 8 +
.../impala/util/ExecutorMembershipSnapshot.java | 6 +-
.../org/apache/impala/planner/ClusterSizeTest.java | 6 +
fe/src/test/resources/fair-scheduler-2-groups.xml | 13 +
fe/src/test/resources/llama-site-2-groups.xml | 36 ++
tests/common/skip.py | 1 +
tests/common/test_dimensions.py | 3 +-
tests/custom_cluster/test_admission_controller.py | 2 +
tests/custom_cluster/test_coordinators.py | 1 +
tests/custom_cluster/test_executor_groups.py | 56 +++
tests/query_test/test_observability.py | 14 +
tests/query_test/test_scanners_fuzz.py | 2 +-
tests/stress/test_acid_stress.py | 5 +-
26 files changed, 679 insertions(+), 66 deletions(-)
create mode 100644 fe/src/test/resources/fair-scheduler-2-groups.xml
create mode 100644 fe/src/test/resources/llama-site-2-groups.xml
[impala] 02/03: IMPALA-11119: [DOCS] document BYTES function
Posted by st...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
stigahuang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit e90cf9c8113ddedd924a807f938d156915fa05d0
Author: Shajini Thayasingh <st...@cloudera.com>
AuthorDate: Thu Feb 24 11:44:49 2022 -0800
IMPALA-11119: [DOCS] document BYTES function
removed white spaces
Change-Id: Iffbe2979f2912cf014866a3c47d88601e04f6e75
Reviewed-on: http://gerrit.cloudera.org:8080/18272
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Reviewed-by: Quanlong Huang <hu...@gmail.com>
---
docs/topics/impala_string_functions.xml | 37 +++++++++++++++++++++++++++++++++
1 file changed, 37 insertions(+)
diff --git a/docs/topics/impala_string_functions.xml b/docs/topics/impala_string_functions.xml
index 438d303..c9fa41c 100644
--- a/docs/topics/impala_string_functions.xml
+++ b/docs/topics/impala_string_functions.xml
@@ -95,6 +95,10 @@ under the License.
</li>
<li>
+ <xref href="#string_functions/bytes">BYTES</xref>
+ </li>
+
+ <li>
<xref href="#string_functions/char_length">CHAR_LENGTH</xref>
</li>
@@ -395,6 +399,39 @@ select concat('[',btrim('xyhelxyzlozyzzxx','xyz'),']');
</dlentry>
+ <dlentry id="bytes">
+ <dt>BYTES</dt>
+ <dd><b>Purpose:</b> Returns the number of bytes contained in the specified byte string. <p>
+ <b>Syntax:</b> BYTES (<i>byte_expression</i>)</p><p>Where:</p><p><i>byte_expression</i>
+ is the byte string for which the number of bytes is to be returned.</p><p
+ conref="../shared/impala_common.xml#common/usage_notes_blurb"/><p>The BYTES function is
+ similar to the LENGTH() function except that it always returns the number of bytes
+ regardless of the status of UTF-8 mode whether it is turned ON or OFF.</p><p>The
+ following is the list of supported string data types to be used in
+ <i>byte_expression</i>:</p><ul>
+ <li>STRING</li>
+ <li>VARCHAR</li>
+ <li>CHAR</li>
+ </ul><note>The BYTES (<i>byte_expression</i>) function counts the trailing zeros because
+ trailing zero bytes are considered bytes. In the case of a fixed <i>length</i> column,
+ the length of the value is always equal to the length defined for the column whereas the
+ length of the value in a <i>variable</i> length column is always equal to the number of
+ bytes, including any trailing zero bytes, contained in that value.</note><p
+ conref="../shared/impala_common.xml#common/example_blurb"/><p>The following example
+ obtains the number of bytes from “cloudera” by applying the BYTES function to the column
+ “cloudera”, which is type
+ VARCHAR.</p><codeblock>SELECT bytes(cast('cloudera' as varchar(20))) FROM xyz;
+ +----------------------------------------+
+ | bytes(cast('cloudera' as varchar(20))) |
+ +----------------------------------------+
+ | 8 |
+ +----------------------------------------+
+ </codeblock><note>Use the TRIM function on the <i>byte_expression</i> to
+ exclude the trailing blanks included in the byte count for a data value.</note>
+ </dd>
+
+ </dlentry>
+
<dlentry rev="1.3.0" id="char_length">
<dt>
[impala] 01/03: IMPALA-10992 Planner changes for estimate peak memory
Posted by st...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
stigahuang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 07a3e6e0df8ed05577a52d896d0ccfbbf6dd8e28
Author: Qifan Chen <qc...@cloudera.com>
AuthorDate: Wed Jan 5 17:20:00 2022 -0500
IMPALA-10992 Planner changes for estimate peak memory
This patch provides replan support for multiple executor group sets.
Each executor group set is associated with a distinct number of nodes
and a threshold for estimated memory per host in bytes that can be
denoted as [<group_name_prefix>:<#nodes>, <threshold>].
In the patch, a query of type EXPLAIN, QUERY or DML can be compiled
more than once. In each attempt, per host memory is estimated and
compared with the threshold of an executor group set. If the estimated
memory is no more than the threshold, the iteration process terminates
and the final plan is determined. The executor group set with the
threshold is selected to run the query.
A new query option 'enable_replan', default to 1 (enabled), is added.
It can be set to 0 to disable this patch and to generate the distributed
plan for the default executor group.
To avoid long compilation time, the following enhancement is enabled.
Note 1) can be disabled when relevant meta-data change is
detected.
1. Authorization is performed only for the 1st compilation;
2. openTransaction() is called for transactional queries in 1st
compilation and the saved transactional info is used in
subsequent compilations. Similar logic is applied to Kudu
transactional queries.
To facilitate testing, the patch imposes an artificial two executor
group setup in FE as follows.
1. [regular:<#nodes>, 64MB]
2. [large:<#nodes>, 8PB]
This setup is enabled when a new query option 'test_replan' is set
to 1 in backend tests, or RuntimeEnv.INSTANCE.isTestEnv() is true as
in most frontend tests. This query option is set to 0 by default.
Compilation time increases when a query is compiled in several
iterations, as shown below for several TPCDs queries. The increase
is mostly due to redundant work in either single node plan creation
or recomputing value transfer graph phase. For small queries, the
increase can be avoided if they can be compiled in single iteration
by properly setting the smallest threshold among all executor group
sets. For example, for the set of queries listed below, the smallest
threshold can be set to 320MB to catch both q15 and q21 in one
compilation.
Compilation time (ms)
Queries Estimated Memory 2-iterations 1-iteration Percentage of
increase
q1 408MB 60.14 25.75 133.56%
q11 1.37GB 261.00 109.61 138.11%
q10a 519MB 139.24 54.52 155.39%
q13 339MB 143.82 60.08 139.38%
q14a 3.56GB 762.68 312.92 143.73%
q14b 2.20GB 522.01 245.13 112.95%
q15 314MB 9.73 4.28 127.33%
q21 275MB 16.00 8.18 95.59%
q23a 1.50GB 461.69 231.78 99.19%
q23b 1.34GB 461.31 219.61 110.05%
q4 2.60GB 218.05 105.07 107.52%
q67 5.16GB 694.59 334.24 101.82%
Testing:
1. Almost all FE and BE tests are now run in the artificial two
executor setup except a few where a specific cluster configuration
is desirable;
2. Ran core tests successfully;
3. Added a new observability test and a new query assignment test;
4. Disabled concurrent insert test (test_concurrent_inserts) and
failing inserts (test_failing_inserts) test in local catalog mode
due to flakiness. Reported both in IMPALA-11189 and IMPALA-11191.
Change-Id: I75cf17290be2c64fd4b732a5505bdac31869712a
Reviewed-on: http://gerrit.cloudera.org:8080/18178
Reviewed-by: Qifan Chen <qc...@cloudera.com>
Tested-by: Qifan Chen <qc...@cloudera.com>
---
be/src/service/query-options.cc | 8 +
be/src/service/query-options.h | 6 +-
common/thrift/Frontend.thrift | 7 +
common/thrift/ImpalaService.thrift | 6 +
common/thrift/Query.thrift | 11 +
.../apache/impala/analysis/AnalysisContext.java | 25 +-
.../java/org/apache/impala/analysis/Analyzer.java | 16 +
.../org/apache/impala/planner/HBaseScanNode.java | 2 +-
.../org/apache/impala/planner/HdfsScanNode.java | 14 +-
.../org/apache/impala/planner/KuduScanNode.java | 9 +-
.../impala/planner/ResourceProfileBuilder.java | 1 +
.../java/org/apache/impala/service/Frontend.java | 450 +++++++++++++++++++--
.../java/org/apache/impala/util/ClassUtil.java | 8 +
.../impala/util/ExecutorMembershipSnapshot.java | 6 +-
.../org/apache/impala/planner/ClusterSizeTest.java | 6 +
fe/src/test/resources/fair-scheduler-2-groups.xml | 13 +
fe/src/test/resources/llama-site-2-groups.xml | 36 ++
tests/common/skip.py | 1 +
tests/common/test_dimensions.py | 3 +-
tests/custom_cluster/test_admission_controller.py | 2 +
tests/custom_cluster/test_coordinators.py | 1 +
tests/custom_cluster/test_executor_groups.py | 56 +++
tests/query_test/test_observability.py | 14 +
tests/stress/test_acid_stress.py | 5 +-
24 files changed, 641 insertions(+), 65 deletions(-)
diff --git a/be/src/service/query-options.cc b/be/src/service/query-options.cc
index 2030f3f..fee9427 100644
--- a/be/src/service/query-options.cc
+++ b/be/src/service/query-options.cc
@@ -1201,6 +1201,14 @@ Status impala::SetQueryOption(const string& key, const string& value,
query_options->__set_runtime_in_list_filter_entry_limit(limit);
break;
}
+ case TImpalaQueryOptions::ENABLE_REPLAN: {
+ query_options->__set_enable_replan(IsTrue(value));
+ break;
+ }
+ case TImpalaQueryOptions::TEST_REPLAN: {
+ query_options->__set_test_replan(IsTrue(value));
+ break;
+ }
default:
if (IsRemovedQueryOption(key)) {
LOG(WARNING) << "Ignoring attempt to set removed query option '" << key << "'";
diff --git a/be/src/service/query-options.h b/be/src/service/query-options.h
index 10ebb77..a31d581 100644
--- a/be/src/service/query-options.h
+++ b/be/src/service/query-options.h
@@ -50,7 +50,7 @@ typedef std::unordered_map<string, beeswax::TQueryOptionLevel::type>
// time we add or remove a query option to/from the enum TImpalaQueryOptions.
#define QUERY_OPTS_TABLE\
DCHECK_EQ(_TImpalaQueryOptions_VALUES_TO_NAMES.size(),\
- TImpalaQueryOptions::RUNTIME_IN_LIST_FILTER_ENTRY_LIMIT + 1);\
+ TImpalaQueryOptions::TEST_REPLAN+ 1);\
REMOVED_QUERY_OPT_FN(abort_on_default_limit_exceeded, ABORT_ON_DEFAULT_LIMIT_EXCEEDED)\
QUERY_OPT_FN(abort_on_error, ABORT_ON_ERROR, TQueryOptionLevel::REGULAR)\
REMOVED_QUERY_OPT_FN(allow_unsupported_formats, ALLOW_UNSUPPORTED_FORMATS)\
@@ -277,6 +277,10 @@ typedef std::unordered_map<string, beeswax::TQueryOptionLevel::type>
QUERY_OPT_FN(orc_async_read, ORC_ASYNC_READ, TQueryOptionLevel::ADVANCED)\
QUERY_OPT_FN(runtime_in_list_filter_entry_limit,\
RUNTIME_IN_LIST_FILTER_ENTRY_LIMIT, TQueryOptionLevel::ADVANCED)\
+ QUERY_OPT_FN(enable_replan, ENABLE_REPLAN,\
+ TQueryOptionLevel::ADVANCED)\
+ QUERY_OPT_FN(test_replan, TEST_REPLAN,\
+ TQueryOptionLevel::ADVANCED)\
;
/// Enforce practical limits on some query options to avoid undesired query state.
diff --git a/common/thrift/Frontend.thrift b/common/thrift/Frontend.thrift
index b2d9a4c..523273e 100644
--- a/common/thrift/Frontend.thrift
+++ b/common/thrift/Frontend.thrift
@@ -725,6 +725,13 @@ struct TExecutorGroupSet {
// Note: this will be empty when 'default' executor group is used or
// 'expected_executor_group_sets' startup flag is not specified.
3: string exec_group_name_prefix
+
+ // The optional max_mem_limit to determine which executor group set to run for a query.
+ // The max_mem_limit value is set to the max_query_mem_limit attribute of the group set
+ // with name prefix 'exec_group_name_prefix' from the pool service. For each query,
+ // the frontend computes the per host estimated-memory after a compilation with a
+ // number of executor nodes from this group set and compares it with this variable.
+ 4: optional i64 max_mem_limit
}
// Sent from the impalad BE to FE with the latest membership snapshot of the
diff --git a/common/thrift/ImpalaService.thrift b/common/thrift/ImpalaService.thrift
index fdb0ec1..212d14d 100644
--- a/common/thrift/ImpalaService.thrift
+++ b/common/thrift/ImpalaService.thrift
@@ -724,6 +724,12 @@ enum TImpalaQueryOptions {
// Maximum number of distinct entries in a runtime in-list filter.
RUNTIME_IN_LIST_FILTER_ENTRY_LIMIT = 142;
+
+ // If true, replanning is enabled.
+ ENABLE_REPLAN = 143;
+
+ // If true, test replan by imposing artificial two executor groups in FE.
+ TEST_REPLAN = 144;
}
// The summary of a DML statement.
diff --git a/common/thrift/Query.thrift b/common/thrift/Query.thrift
index 3e8b0b9..f868bdf 100644
--- a/common/thrift/Query.thrift
+++ b/common/thrift/Query.thrift
@@ -577,6 +577,17 @@ struct TQueryOptions {
// See comment in ImpalaService.thrift
143: optional i32 runtime_in_list_filter_entry_limit = 1024;
+
+ // Indicates whether to enable auto-scaling which is a process to generate a suitable
+ // plan among different-sized executor group sets. The returned plan satisfies the
+ // resource requirement imposed on the executor group set. Default is to enable.
+ 144: optional bool enable_replan = true;
+
+ // Set to true to programmatically treat the default executor group as a two-executor
+ // groups in FE as follows.
+ // 1. regular: <num_nodes> nodes with 64MB of per-host estimated memory threshold
+ // 2. large: <num_nodes> nodes with 8PB of per-host estimated memory threshold
+ 145: optional bool test_replan = false;
}
// Impala currently has three types of sessions: Beeswax, HiveServer2 and external
diff --git a/fe/src/main/java/org/apache/impala/analysis/AnalysisContext.java b/fe/src/main/java/org/apache/impala/analysis/AnalysisContext.java
index f0c7fe9..078d56a 100644
--- a/fe/src/main/java/org/apache/impala/analysis/AnalysisContext.java
+++ b/fe/src/main/java/org/apache/impala/analysis/AnalysisContext.java
@@ -435,6 +435,13 @@ public class AnalysisContext {
return result;
}
+ public AnalysisResult analyzeAndAuthorize(StatementBase stmt,
+ StmtTableCache stmtTableCache, AuthorizationChecker authzChecker)
+ throws ImpalaException {
+ return analyzeAndAuthorize(
+ stmt, stmtTableCache, authzChecker, false /*disableAuthorization*/);
+ }
+
/**
* Analyzes and authorizes the given statement using the provided table cache and
* authorization checker.
@@ -442,8 +449,8 @@ public class AnalysisContext {
* reveal the existence/absence of objects the user is not authorized to see.
*/
public AnalysisResult analyzeAndAuthorize(StatementBase stmt,
- StmtTableCache stmtTableCache, AuthorizationChecker authzChecker)
- throws ImpalaException {
+ StmtTableCache stmtTableCache, AuthorizationChecker authzChecker,
+ boolean disableAuthorization) throws ImpalaException {
// TODO: Clean up the creation/setting of the analysis result.
analysisResult_ = new AnalysisResult();
analysisResult_.stmt_ = stmt;
@@ -469,12 +476,14 @@ public class AnalysisContext {
// Authorize statement and record exception. Authorization relies on information
// collected during analysis.
AuthorizationException authException = null;
- try {
- authzChecker.authorize(authzCtx, analysisResult_, catalog_);
- } catch (AuthorizationException e) {
- authException = e;
- } finally {
- authzChecker.postAuthorize(authzCtx, authException == null);
+ if (!disableAuthorization) {
+ try {
+ authzChecker.authorize(authzCtx, analysisResult_, catalog_);
+ } catch (AuthorizationException e) {
+ authException = e;
+ } finally {
+ authzChecker.postAuthorize(authzCtx, authException == null);
+ }
}
// AuthorizationExceptions take precedence over AnalysisExceptions so as not
diff --git a/fe/src/main/java/org/apache/impala/analysis/Analyzer.java b/fe/src/main/java/org/apache/impala/analysis/Analyzer.java
index b51a22b..d8d4126 100644
--- a/fe/src/main/java/org/apache/impala/analysis/Analyzer.java
+++ b/fe/src/main/java/org/apache/impala/analysis/Analyzer.java
@@ -95,6 +95,7 @@ import org.apache.impala.thrift.TQueryCtx;
import org.apache.impala.thrift.TQueryOptions;
import org.apache.impala.util.AcidUtils;
import org.apache.impala.util.DisjointSet;
+import org.apache.impala.util.ExecutorMembershipSnapshot;
import org.apache.impala.util.Graph.RandomAccessibleGraph;
import org.apache.impala.util.Graph.SccCondensedGraph;
import org.apache.impala.util.Graph.WritableGraph;
@@ -501,6 +502,12 @@ public class Analyzer {
// require analysis (e.g. some literal expressions).
private int numStmtExprs_ = 0;
+ // When set to a value > -1, it represents the number of executors for current
+ // iteration of compilation. Used by auto scaling to pass the info to the planner in
+ // place of ExecutorMembershipSnapshot.numExecutors(). ExecutorMembershipSnapshot is a
+ // singleton.
+ private int numExecutorsForPlanning_ = -1;
+
// Cache of KuduTables opened for this query. (map from table name to kudu table)
// This cache prevent multiple openTable calls for a given table in the same query.
public final Map<String, org.apache.kudu.client.KuduTable> kuduTables =
@@ -553,6 +560,15 @@ public class Analyzer {
public boolean containsSubquery() { return globalState_.containsSubquery; }
public boolean containsSetOperation() { return globalState_.setOperationNeedsRewrite; }
+ public int numExecutorsForPlanning() {
+ return (globalState_.numExecutorsForPlanning_ >= 0) ?
+ globalState_.numExecutorsForPlanning_ :
+ ExecutorMembershipSnapshot.getCluster().numExecutors();
+ }
+ public void setNumExecutorsForPlanning(int x) {
+ globalState_.numExecutorsForPlanning_ = x;
+ }
+
// An analyzer stores analysis state for a single select block. A select block can be
// a top level select statement, or an inline view select block.
// ancestors contains the Analyzers of the enclosing select blocks of 'this'
diff --git a/fe/src/main/java/org/apache/impala/planner/HBaseScanNode.java b/fe/src/main/java/org/apache/impala/planner/HBaseScanNode.java
index 12c14a3..97a8a42 100644
--- a/fe/src/main/java/org/apache/impala/planner/HBaseScanNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/HBaseScanNode.java
@@ -363,7 +363,7 @@ public class HBaseScanNode extends ScanNode {
// Assume that each node/instance in the cluster gets a scan range, unless there are
// fewer scan ranges than nodes/instances.
- int numExecutors = ExecutorMembershipSnapshot.getCluster().numExecutors();
+ int numExecutors = analyzer.numExecutorsForPlanning();
numNodes_ =
Math.max(1, Math.min(scanRangeSpecs_.getConcrete_rangesSize(), numExecutors));
int maxInstances = numNodes_ * getMaxInstancesPerNode(analyzer);
diff --git a/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java b/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java
index e43d819..a0db795 100644
--- a/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java
@@ -1619,7 +1619,8 @@ public class HdfsScanNode extends ScanNode {
Preconditions.checkNotNull(scanRangeSpecs_);
ExecutorMembershipSnapshot cluster = ExecutorMembershipSnapshot.getCluster();
final int maxInstancesPerNode = getMaxInstancesPerNode(analyzer);
- final int maxPossibleInstances = cluster.numExecutors() * maxInstancesPerNode;
+ final int maxPossibleInstances =
+ analyzer.numExecutorsForPlanning() * maxInstancesPerNode;
int totalNodes = 0;
int totalInstances = 0;
int numLocalRanges = 0;
@@ -1683,10 +1684,12 @@ public class HdfsScanNode extends ScanNode {
// hosts that hold block replica for those ranges.
int numLocalNodes = Math.min(numLocalRanges, localRangeCounts.size());
// The remote ranges are round-robined across all the impalads.
- int numRemoteNodes = Math.min(numRemoteRanges, cluster.numExecutors());
+ int numRemoteNodes =
+ Math.min(numRemoteRanges, analyzer.numExecutorsForPlanning());
// The local and remote assignments may overlap, but we don't know by how much
// so conservatively assume no overlap.
- totalNodes = Math.min(numLocalNodes + numRemoteNodes, cluster.numExecutors());
+ totalNodes = Math.min(
+ numLocalNodes + numRemoteNodes, analyzer.numExecutorsForPlanning());
totalInstances = computeNumInstances(numLocalRanges, numRemoteRanges,
totalNodes, maxInstancesPerNode, totalLocalParallelism);
// Exit early if we have maxed out our estimate of hosts/instances, to avoid
@@ -1703,7 +1706,8 @@ public class HdfsScanNode extends ScanNode {
generatedScanRangeCount_ >= scanRangeSpecs_.getSplit_specsSize());
numRemoteRanges += generatedScanRangeCount_;
int numLocalNodes = Math.min(numLocalRanges, localRangeCounts.size());
- totalNodes = Math.min(numLocalNodes + numRemoteRanges, cluster.numExecutors());
+ totalNodes =
+ Math.min(numLocalNodes + numRemoteRanges, analyzer.numExecutorsForPlanning());
totalInstances = computeNumInstances(numLocalRanges, numRemoteRanges,
totalNodes, maxInstancesPerNode, totalLocalParallelism);
}
@@ -1717,7 +1721,7 @@ public class HdfsScanNode extends ScanNode {
+ " localRanges=" + numLocalRanges + " remoteRanges=" + numRemoteRanges
+ " localRangeCounts.size=" + localRangeCounts.size()
+ " totalLocalParallelism=" + totalLocalParallelism
- + " executorNodes=" + cluster.numExecutors() + " "
+ + " executorNodes=" + analyzer.numExecutorsForPlanning() + " "
+ " numNodes=" + numNodes_ + " numInstances=" + numInstances_);
}
}
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 dbb49a7..d0d106a 100644
--- a/fe/src/main/java/org/apache/impala/planner/KuduScanNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/KuduScanNode.java
@@ -304,7 +304,8 @@ public class KuduScanNode extends ScanNode {
protected void computeNumNodes(Analyzer analyzer) {
ExecutorMembershipSnapshot cluster = ExecutorMembershipSnapshot.getCluster();
final int maxInstancesPerNode = getMaxInstancesPerNode(analyzer);
- final int maxPossibleInstances = cluster.numExecutors() * maxInstancesPerNode;
+ final int maxPossibleInstances =
+ analyzer.numExecutorsForPlanning() * maxInstancesPerNode;
int totalNodes = 0;
int totalInstances = 0;
int numLocalRanges = 0;
@@ -347,10 +348,12 @@ public class KuduScanNode extends ScanNode {
// hosts that hold replica for those ranges.
int numLocalNodes = Math.min(numLocalRanges, localRangeCounts.size());
// The remote ranges are round-robined across all the impalads.
- int numRemoteNodes = Math.min(numRemoteRanges, cluster.numExecutors());
+ int numRemoteNodes =
+ Math.min(numRemoteRanges, analyzer.numExecutorsForPlanning());
// The local and remote assignments may overlap, but we don't know by how much
// so conservatively assume no overlap.
- totalNodes = Math.min(numLocalNodes + numRemoteNodes, cluster.numExecutors());
+ totalNodes =
+ Math.min(numLocalNodes + numRemoteNodes, analyzer.numExecutorsForPlanning());
int numLocalInstances = Math.min(numLocalRanges, totalLocalParallelism);
totalInstances = Math.min(numLocalInstances + numRemoteRanges,
diff --git a/fe/src/main/java/org/apache/impala/planner/ResourceProfileBuilder.java b/fe/src/main/java/org/apache/impala/planner/ResourceProfileBuilder.java
index 60c231b..1467aad 100644
--- a/fe/src/main/java/org/apache/impala/planner/ResourceProfileBuilder.java
+++ b/fe/src/main/java/org/apache/impala/planner/ResourceProfileBuilder.java
@@ -17,6 +17,7 @@
package org.apache.impala.planner;
+import org.apache.impala.planner.ResourceProfile;
import com.google.common.base.Preconditions;
/**
diff --git a/fe/src/main/java/org/apache/impala/service/Frontend.java b/fe/src/main/java/org/apache/impala/service/Frontend.java
index a921bb9..ced9574 100644
--- a/fe/src/main/java/org/apache/impala/service/Frontend.java
+++ b/fe/src/main/java/org/apache/impala/service/Frontend.java
@@ -121,6 +121,9 @@ import org.apache.impala.common.ImpalaException;
import org.apache.impala.common.InternalException;
import org.apache.impala.common.KuduTransactionManager;
import org.apache.impala.common.NotImplementedException;
+import org.apache.impala.common.Pair;
+import org.apache.impala.common.PrintUtils;
+import org.apache.impala.common.RuntimeEnv;
import org.apache.impala.common.TransactionException;
import org.apache.impala.common.TransactionKeepalive;
import org.apache.impala.common.TransactionKeepalive.HeartbeatContext;
@@ -142,11 +145,13 @@ import org.apache.impala.thrift.TColumn;
import org.apache.impala.thrift.TColumnValue;
import org.apache.impala.thrift.TCommentOnParams;
import org.apache.impala.thrift.TCreateDropRoleParams;
+import org.apache.impala.thrift.TDataSink;
import org.apache.impala.thrift.TDdlExecRequest;
import org.apache.impala.thrift.TDdlType;
import org.apache.impala.thrift.TDescribeOutputStyle;
import org.apache.impala.thrift.TDescribeResult;
import org.apache.impala.thrift.TExecRequest;
+import org.apache.impala.thrift.TExecutorGroupSet;
import org.apache.impala.thrift.TExplainResult;
import org.apache.impala.thrift.TFinalizeParams;
import org.apache.impala.thrift.TFunctionCategory;
@@ -155,6 +160,8 @@ import org.apache.impala.thrift.TGetTableHistoryResult;
import org.apache.impala.thrift.TGetTableHistoryResultItem;
import org.apache.impala.thrift.TGrantRevokePrivParams;
import org.apache.impala.thrift.TGrantRevokeRoleParams;
+import org.apache.impala.thrift.THdfsTableSink;
+import org.apache.impala.thrift.TKuduTableSink;
import org.apache.impala.thrift.TLineageGraph;
import org.apache.impala.thrift.TLoadDataReq;
import org.apache.impala.thrift.TLoadDataResp;
@@ -162,6 +169,7 @@ import org.apache.impala.thrift.TCopyTestCaseReq;
import org.apache.impala.thrift.TMetadataOpRequest;
import org.apache.impala.thrift.TPlanExecInfo;
import org.apache.impala.thrift.TPlanFragment;
+import org.apache.impala.thrift.TPoolConfig;
import org.apache.impala.thrift.TQueryCtx;
import org.apache.impala.thrift.TQueryExecRequest;
import org.apache.impala.thrift.TQueryOptions;
@@ -173,6 +181,7 @@ import org.apache.impala.thrift.TShowFilesParams;
import org.apache.impala.thrift.TShowStatsOp;
import org.apache.impala.thrift.TStmtType;
import org.apache.impala.thrift.TTableName;
+import org.apache.impala.thrift.TTableSink;
import org.apache.impala.thrift.TTruncateParams;
import org.apache.impala.thrift.TUniqueId;
import org.apache.impala.thrift.TUpdateCatalogCacheRequest;
@@ -184,6 +193,7 @@ import org.apache.impala.util.ExecutorMembershipSnapshot;
import org.apache.impala.util.IcebergUtil;
import org.apache.impala.util.KuduUtil;
import org.apache.impala.util.PatternMatcher;
+import org.apache.impala.util.RequestPoolService;
import org.apache.impala.util.TResultRowBuilder;
import org.apache.impala.util.TSessionStateUtil;
import org.apache.kudu.client.KuduClient;
@@ -202,6 +212,9 @@ import com.google.common.collect.Sets;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.Uninterruptibles;
+import static org.apache.impala.common.ByteUnits.MEGABYTE;
+import static org.apache.impala.common.ByteUnits.GIGABYTE;
+
/**
* Frontend API for the impalad process.
* This class allows the impala daemon to create TQueryExecRequest
@@ -225,6 +238,10 @@ public class Frontend {
// show tables/databases.
private static final int MAX_CHECK_AUTHORIZATION_POOL_SIZE = 128;
+ // The default pool name to use when ExecutorMembershipSnapshot is in initial state
+ // (i.e., ExecutorMembershipSnapshot.numExecutors_ == 0).
+ private static final String DEFAULT_POOL_NAME = "default-pool";
+
/**
* Plan-time context that allows capturing various artifacts created
* during the process.
@@ -249,14 +266,109 @@ public class Frontend {
// Thrift. For unit testing.
protected List<PlanFragment> plan_;
+ // An inner class to capture the state of compilation for auto-scaling.
+ final class AutoScalingCompilationState {
+ // Flag to indicate whether to disable authorization after analyze. Used by
+ // auto-scalling to avoid redundent authorizations.
+ protected boolean disableAuthorization_ = false;
+
+ // The estimated memory per host for certain queries that do not populate
+ // TExecRequest.query_exec_request field.
+ protected long estimated_memory_per_host_ = -1;
+
+ // The initial length of content in explain buffer to help return the buffer
+ // to the initial position prior to another auto-scaling compilation.
+ protected int initialExplainBufLen_ = -1;
+
+ // The initial query options seen before any compilations, which will be copied
+ // and used by each iteration of auto-scaling compilation.
+ protected TQueryOptions initialQueryOptions_ = null;
+
+ // The allocated write Id when in an ACID transaction.
+ protected long writeId_ = -1;
+
+ // The transaction token when in a Kudu transaction.
+ protected byte[] kuduTransactionToken_ = null;
+
+ // Indicate whether runtime profile/summary can be accessed. Set at the end of
+ // 1st iteration.
+ protected boolean user_has_profile_access_ = false;
+
+ // The group set being applied in current compilation.
+ protected TExecutorGroupSet group_set_ = null;
+
+ public boolean disableAuthorization() { return disableAuthorization_; }
+
+ public long getEstimatedMemoryPerHost() { return estimated_memory_per_host_; }
+ public void setEstimatedMemoryPerHost(long x) { estimated_memory_per_host_ = x; }
+
+ // Capture the current state and initialize before iterative compilations begin.
+ public void captureState() {
+ disableAuthorization_ = false;
+ estimated_memory_per_host_ = -1;
+ initialExplainBufLen_ = PlanCtx.this.explainBuf_.length();
+ initialQueryOptions_ =
+ new TQueryOptions(getQueryContext().client_request.getQuery_options());
+ writeId_ = -1;
+ kuduTransactionToken_ = null;
+ }
+
+ // Restore to the captured state after an iterative compilation
+ public void restoreState() {
+ // Avoid authorization starting from 2nd iteration. This flag can be set to
+ // false when meta-data change is detected.
+ disableAuthorization_ = true;
+
+ // Reset estimated memory
+ estimated_memory_per_host_ = -1;
+
+ // Remove the explain string accumulated
+ explainBuf_.delete(initialExplainBufLen_, explainBuf_.length());
+
+ // Use a brand new copy of query options
+ getQueryContext().client_request.setQuery_options(
+ new TQueryOptions(initialQueryOptions_));
+
+ // Set the flag to false to avoid serializing TQueryCtx.desc_tbl_testonly (a list
+ // of Descriptors.TDescriptorTable) in next iteration of compilation.
+ // TQueryCtx.desc_tbl_testonly is set in current iteration during planner test.
+ queryCtx_.setDesc_tbl_testonlyIsSet(false);
+ }
+
+ // When exception InconsistentMetadataFetchException is received and before
+ // next compilation, disable stmt cache and re-authorize.
+ public void disableStmtCacheAndReauthorize() {
+ restoreState();
+ disableAuthorization_ = false;
+ }
+
+ long getWriteId() { return writeId_; }
+ void setWriteId(long x) { writeId_ = x; }
+
+ byte[] getKuduTransactionToken() { return kuduTransactionToken_; }
+ void setKuduTransactionToken(byte[] token) {
+ kuduTransactionToken_ = (token == null) ? null : token.clone();
+ }
+
+ boolean userHasProfileAccess() { return user_has_profile_access_; }
+ void setUserHasProfileAccess(boolean x) { user_has_profile_access_ = x; }
+
+ TExecutorGroupSet getGroupSet() { return group_set_; }
+ void setGroupSet(TExecutorGroupSet x) { group_set_ = x; }
+ }
+
+ public AutoScalingCompilationState compilationState_;
+
public PlanCtx(TQueryCtx qCtx) {
queryCtx_ = qCtx;
explainBuf_ = new StringBuilder();
+ compilationState_ = new AutoScalingCompilationState();
}
public PlanCtx(TQueryCtx qCtx, StringBuilder describe) {
queryCtx_ = qCtx;
explainBuf_ = describe;
+ compilationState_ = new AutoScalingCompilationState();
}
/**
@@ -1597,6 +1709,10 @@ public class Frontend {
planCtx.explainBuf_.append(planner.getExplainString(allFragments, result));
result.setQuery_plan(planCtx.getExplainString());
+ // copy estimated memory per host to planCtx for auto-scaling.
+ planCtx.compilationState_.setEstimatedMemoryPerHost(
+ result.getPer_host_mem_estimate());
+
return result;
}
@@ -1630,34 +1746,238 @@ public class Frontend {
+ "%s of %s times: ",numRetries, INCONSISTENT_METADATA_NUM_RETRIES) + msg);
}
+ /**
+ * Examines the input 'executorGroupSets', removes non-default and useless group sets
+ * from it and fills the max_mem_limit field. A group set is considered useless if its
+ * name is not a suffix of 'request_pool'. The max_mem_limit is set to the
+ * max_query_mem_limit from the pool service for the surviving group sets.
+ *
+ * Also imposes the artificial two-executor groups for testing when needed.
+ */
+ private List<TExecutorGroupSet> setupThresholdsForExecutorGroupSets(
+ List<TExecutorGroupSet> executorGroupSets, String request_pool,
+ boolean default_executor_group, boolean test_replan) throws ImpalaException {
+ RequestPoolService poolService = RequestPoolService.getInstance();
+
+ List<TExecutorGroupSet> result = Lists.newArrayList();
+ if (default_executor_group) {
+ TExecutorGroupSet e = executorGroupSets.get(0);
+ if (e.getCurr_num_executors() == 0) {
+ // The default group has 0 executors. Return one with the number of default
+ // executors as the number of expected executors.
+ result.add(new TExecutorGroupSet(e));
+ result.get(0).setCurr_num_executors(e.getExpected_num_executors());
+ result.get(0).setMax_mem_limit(Long.MAX_VALUE);
+ } else if (test_replan) {
+ ExecutorMembershipSnapshot cluster = ExecutorMembershipSnapshot.getCluster();
+ int num_nodes = cluster.numExecutors();
+ Preconditions.checkState(e.getCurr_num_executors() == num_nodes);
+ // Form a two-executor group testing environment so that we can exercise
+ // auto-scaling logic (see getTExecRequest()).
+ TExecutorGroupSet s = new TExecutorGroupSet(e);
+ s.setExec_group_name_prefix("small");
+ s.setMax_mem_limit(64*MEGABYTE);
+ result.add(s);
+ TExecutorGroupSet l = new TExecutorGroupSet(e);
+ String newName = "large";
+ if (e.isSetExec_group_name_prefix()) {
+ String currentName = e.getExec_group_name_prefix();
+ if (currentName.length() > 0) newName = currentName;
+ }
+ l.setExec_group_name_prefix(newName);
+ l.setMax_mem_limit(Long.MAX_VALUE);
+ result.add(l);
+ } else {
+ // Copy and augment the group with the maximally allowed max_mem_limit value.
+ result.add(new TExecutorGroupSet(e));
+ result.get(0).setMax_mem_limit(Long.MAX_VALUE);
+ }
+ return result;
+ }
+
+ // If there are no executor groups in the cluster, Create a group set with 1 executor.
+ if (executorGroupSets.size() == 0) {
+ result.add(new TExecutorGroupSet(1, 20, DEFAULT_POOL_NAME));
+ result.get(0).setMax_mem_limit(Long.MAX_VALUE);
+ return result;
+ }
+
+ // Executor groups exist in the cluster. Identify those that can be used.
+ for (TExecutorGroupSet e : executorGroupSets) {
+ // If defined, request_pool can be a suffix of the group name prefix. For example
+ // group_set_prefix = root.queue1
+ // request_pool = queue1
+ if (request_pool != null && !e.getExec_group_name_prefix().endsWith(request_pool)) {
+ continue;
+ }
+ TExecutorGroupSet new_entry = new TExecutorGroupSet(e);
+ if (poolService != null) {
+ // Find out the max_mem_limit from the pool service
+ TPoolConfig poolConfig =
+ poolService.getPoolConfig(e.getExec_group_name_prefix());
+ Preconditions.checkNotNull(poolConfig);
+ new_entry.setMax_mem_limit(poolConfig.getMax_query_mem_limit());
+ } else {
+ // Set to max possible thresold value when there is no pool service
+ new_entry.setMax_mem_limit(Long.MAX_VALUE);
+ }
+ result.add(new_entry);
+ }
+ if (executorGroupSets.size() > 0 && result.size() == 0 && request_pool != null) {
+ throw new AnalysisException("Request pool: " + request_pool
+ + " does not map to any known executor group set.");
+ }
+
+ // Sort 'executorGroupSets' by max_mem_limit field in ascending order. Use
+ // exec_group_name_prefix to break the tie.
+ Collections.sort(result, new Comparator<TExecutorGroupSet>() {
+ @Override
+ public int compare(TExecutorGroupSet e1, TExecutorGroupSet e2) {
+ int i = Long.compare(e1.getMax_mem_limit(), e2.getMax_mem_limit());
+ if (i == 0) {
+ i = e1.getExec_group_name_prefix().compareTo(e2.getExec_group_name_prefix());
+ }
+ return i;
+ }
+ });
+ return result;
+ }
+
+ // Only the following types of statements are considered auto scalable since each
+ // can be planned by the distributed planner utilizing the number of executors in
+ // an executor group as input.
+ private boolean canStmtBeAutoScaled(TStmtType type) {
+ return type == TStmtType.EXPLAIN || type == TStmtType.QUERY || type == TStmtType.DML;
+ }
+
private TExecRequest getTExecRequest(PlanCtx planCtx, EventSequence timeline)
throws ImpalaException {
TQueryCtx queryCtx = planCtx.getQueryContext();
LOG.info("Analyzing query: " + queryCtx.client_request.stmt + " db: "
+ queryCtx.session.database);
+ TQueryOptions queryOptions = queryCtx.client_request.getQuery_options();
+ boolean enable_replan = queryOptions.isEnable_replan();
+
+ List<TExecutorGroupSet> originalExecutorGroupSets =
+ ExecutorMembershipSnapshot.getAllExecutorGroupSets();
+
+ LOG.info("The original executor group sets from executor membership snapshot: "
+ + originalExecutorGroupSets);
+
+ boolean default_executor_group = false;
+ if (originalExecutorGroupSets.size() == 1) {
+ TExecutorGroupSet e = originalExecutorGroupSets.get(0);
+ default_executor_group = e.getExec_group_name_prefix() == null
+ || e.getExec_group_name_prefix().isEmpty();
+ }
+
+ List<TExecutorGroupSet> executorGroupSetsToUse = setupThresholdsForExecutorGroupSets(
+ originalExecutorGroupSets, queryOptions.getRequest_pool(), default_executor_group,
+ enable_replan
+ && (RuntimeEnv.INSTANCE.isTestEnv() || queryOptions.isTest_replan()));
+
+ int num_executor_group_sets = executorGroupSetsToUse.size();
+ if (num_executor_group_sets == 0) {
+ throw new AnalysisException(
+ "No suitable executor group sets can be identified and used.");
+ }
+ LOG.info("A total of {} executor group sets to be considered for auto-scaling: "
+ + executorGroupSetsToUse, num_executor_group_sets);
+
+ TExecRequest req = null;
+
+ // Capture the current state.
+ planCtx.compilationState_.captureState();
+
+ TExecutorGroupSet group_set = null;
+ long per_host_mem_estimate = -1;
+ String reason = "Unknown";
int attempt = 0;
- String retryMsg = "";
- while (true) {
- try {
- TExecRequest req = doCreateExecRequest(planCtx, timeline);
- markTimelineRetries(attempt, retryMsg, timeline);
- return req;
- } catch (InconsistentMetadataFetchException e) {
- if (attempt++ == INCONSISTENT_METADATA_NUM_RETRIES) {
- markTimelineRetries(attempt, e.getMessage(), timeline);
- throw e;
+ for (int i = 0; i < num_executor_group_sets; i++) {
+ group_set = executorGroupSetsToUse.get(i);
+ planCtx.compilationState_.setGroupSet(group_set);
+ LOG.info("Consider executor group set: " + group_set);
+
+ String retryMsg = "";
+ while (true) {
+ try {
+ req = doCreateExecRequest(planCtx, timeline);
+ markTimelineRetries(attempt, retryMsg, timeline);
+ break;
+ } catch (InconsistentMetadataFetchException e) {
+ if (attempt++ == INCONSISTENT_METADATA_NUM_RETRIES) {
+ markTimelineRetries(attempt, e.getMessage(), timeline);
+ throw e;
+ }
+ planCtx.compilationState_.disableStmtCacheAndReauthorize();
+ if (attempt > 1) {
+ // Back-off a bit on later retries.
+ Uninterruptibles.sleepUninterruptibly(200 * attempt, TimeUnit.MILLISECONDS);
+ }
+ retryMsg = e.getMessage();
+ LOG.warn("Retrying plan of query {}: {} (retry #{} of {})",
+ queryCtx.client_request.stmt, retryMsg, attempt,
+ INCONSISTENT_METADATA_NUM_RETRIES);
}
- if (attempt > 1) {
- // Back-off a bit on later retries.
- Uninterruptibles.sleepUninterruptibly(200 * attempt, TimeUnit.MILLISECONDS);
+ }
+
+ // If it is for a single node plan, enable_replan is disabled, or it is not a query
+ // that can be auto scaled, return the 1st plan generated.
+ if (queryOptions.num_nodes == 1) {
+ reason = "the number of nodes is 1";
+ break;
+ } else if (!enable_replan) {
+ reason = "query option 'enable_replan' is false";
+ break;
+ } else if (!canStmtBeAutoScaled(req.stmt_type)) {
+ reason = "query is not auto-scalable";
+ break;
+ }
+
+ // Find out the per host memory estimated from two possible sources.
+ per_host_mem_estimate = -1;
+ if (req.query_exec_request != null) {
+ // For non-explain queries
+ per_host_mem_estimate = req.query_exec_request.per_host_mem_estimate;
+ } else {
+ // For explain queries
+ per_host_mem_estimate = planCtx.compilationState_.getEstimatedMemoryPerHost();
+ }
+
+ Preconditions.checkState(per_host_mem_estimate >= 0);
+
+ if (per_host_mem_estimate <= group_set.getMax_mem_limit()) {
+ reason = "suitable group found (estimated per-host memory="
+ + PrintUtils.printBytes(per_host_mem_estimate) + ")";
+
+ // Set the group name prefix in both the returned query options and
+ // the query context for non default group setup.
+ if (!default_executor_group) {
+ String namePrefix = group_set.getExec_group_name_prefix();
+ req.query_options.setRequest_pool(namePrefix);
+ if (req.query_exec_request != null) {
+ req.query_exec_request.query_ctx.setRequest_pool(namePrefix);
+ }
}
- retryMsg = e.getMessage();
- LOG.warn("Retrying plan of query {}: {} (retry #{} of {})",
- queryCtx.client_request.stmt, retryMsg, attempt,
- INCONSISTENT_METADATA_NUM_RETRIES);
+
+ break;
}
+
+ // Restore to the captured state.
+ planCtx.compilationState_.restoreState();
}
+
+ if (reason.equals("Unknown") && group_set.getMax_mem_limit() > 0) {
+ throw new AnalysisException("The query does not fit any executor group sets.");
+ }
+
+ LOG.info("Selected executor group: " + group_set + ", reason: " + reason);
+
+ // Transfer the profile access flag which is collected during 1st compilation.
+ req.setUser_has_profile_access(planCtx.compilationState_.userHasProfileAccess());
+
+ return req;
}
private TExecRequest doCreateExecRequest(PlanCtx planCtx,
@@ -1675,11 +1995,23 @@ public class Frontend {
// Analyze and authorize stmt
AnalysisContext analysisCtx = new AnalysisContext(queryCtx, authzFactory_, timeline);
AnalysisResult analysisResult = analysisCtx.analyzeAndAuthorize(stmt, stmtTableCache,
- authzChecker_.get());
- LOG.info("Analysis and authorization finished.");
+ authzChecker_.get(), planCtx.compilationState_.disableAuthorization());
+ if (!planCtx.compilationState_.disableAuthorization()) {
+ LOG.info("Analysis and authorization finished.");
+ planCtx.compilationState_.setUserHasProfileAccess(
+ analysisResult.userHasProfileAccess());
+ } else {
+ LOG.info("Analysis finished.");
+ }
Preconditions.checkNotNull(analysisResult.getStmt());
TExecRequest result = createBaseExecRequest(queryCtx, analysisResult);
+ // Transfer the current number of executors in executor group set from planCtx to
+ // analyzer's global state. The info is needed to compute the number of nodes to be
+ // used during planner phase for scans (see HdfsScanNode.computeNumNodes()).
+ analysisResult.getAnalyzer().setNumExecutorsForPlanning(
+ planCtx.compilationState_.getGroupSet().getCurr_num_executors());
+
try {
TQueryOptions queryOptions = queryCtx.client_request.query_options;
if (analysisResult.isCatalogOp()) {
@@ -1701,18 +2033,27 @@ public class Frontend {
FeTable targetTable = insertStmt.getTargetTable();
if (AcidUtils.isTransactionalTable(
targetTable.getMetaStoreTable().getParameters())) {
- long txnId = openTransaction(queryCtx);
- timeline.markEvent("Transaction opened (" + String.valueOf(txnId) + ")");
- Collection<FeTable> tables = stmtTableCache.tables.values();
- String staticPartitionTarget = null;
- if (insertStmt.isStaticPartitionTarget()) {
- staticPartitionTarget = FeCatalogUtils.getPartitionName(
- insertStmt.getPartitionKeyValues());
+
+ if (planCtx.compilationState_.getWriteId() == -1) {
+ // 1st time compilation. Open a transaction and save the writeId.
+ long txnId = openTransaction(queryCtx);
+ timeline.markEvent("Transaction opened (" + String.valueOf(txnId) + ")");
+ Collection<FeTable> tables = stmtTableCache.tables.values();
+ String staticPartitionTarget = null;
+ if (insertStmt.isStaticPartitionTarget()) {
+ staticPartitionTarget = FeCatalogUtils.getPartitionName(
+ insertStmt.getPartitionKeyValues());
+ }
+ createLockForInsert(txnId, tables, targetTable, insertStmt.isOverwrite(),
+ staticPartitionTarget);
+ long writeId = allocateWriteId(queryCtx, targetTable);
+ insertStmt.setWriteId(writeId);
+
+ planCtx.compilationState_.setWriteId(writeId);
+ } else {
+ // Continue the transaction by reusing the writeId.
+ insertStmt.setWriteId(planCtx.compilationState_.getWriteId());
}
- createLockForInsert(txnId, tables, targetTable, insertStmt.isOverwrite(),
- staticPartitionTarget);
- long writeId = allocateWriteId(queryCtx, targetTable);
- insertStmt.setWriteId(writeId);
}
} else if (analysisResult.isLoadDataStmt()) {
result.stmt_type = TStmtType.LOAD;
@@ -1749,23 +2090,23 @@ public class Frontend {
return result;
}
- // Open Kudu transaction if Kudu transaction is enabled and target table is Kudu
- // table.
+ // Open or continue Kudu transaction if Kudu transaction is enabled and target table
+ // is Kudu table.
if (!analysisResult.isExplainStmt() && queryOptions.isEnable_kudu_transaction()) {
if ((analysisResult.isInsertStmt() || analysisResult.isCreateTableAsSelectStmt())
&& analysisResult.getInsertStmt().isTargetTableKuduTable()) {
- // Open Kudu transaction for INSERT/UPSERT/CTAS statements.
- openKuduTransaction(queryCtx, analysisResult,
+ // For INSERT/UPSERT/CTAS statements.
+ openOrContinueKuduTransaction(planCtx, queryCtx, analysisResult,
analysisResult.getInsertStmt().getTargetTable(), timeline);
} else if (analysisResult.isUpdateStmt()
&& analysisResult.getUpdateStmt().isTargetTableKuduTable()) {
- // Open Kudu transaction for UPDATE statement.
- openKuduTransaction(queryCtx, analysisResult,
+ // For UPDATE statement.
+ openOrContinueKuduTransaction(planCtx, queryCtx, analysisResult,
analysisResult.getUpdateStmt().getTargetTable(), timeline);
} else if (analysisResult.isDeleteStmt()
&& analysisResult.getDeleteStmt().isTargetTableKuduTable()) {
- // Open Kudu transaction for DELETE statement.
- openKuduTransaction(queryCtx, analysisResult,
+ // For DELETE statement.
+ openOrContinueKuduTransaction(planCtx, queryCtx, analysisResult,
analysisResult.getDeleteStmt().getTargetTable(), timeline);
}
}
@@ -1819,6 +2160,7 @@ public class Frontend {
} catch (Exception e) {
if (queryCtx.isSetTransaction_id()) {
try {
+ planCtx.compilationState_.setWriteId(-1);
abortTransaction(queryCtx.getTransaction_id());
timeline.markEvent("Transaction aborted");
} catch (TransactionException te) {
@@ -1826,6 +2168,7 @@ public class Frontend {
}
} else if (queryCtx.isIs_kudu_transactional()) {
try {
+ planCtx.compilationState_.setKuduTransactionToken(null);
abortKuduTransaction(queryCtx.getQuery_id());
timeline.markEvent(
"Kudu transaction aborted: " + queryCtx.getQuery_id().toString());
@@ -2233,10 +2576,25 @@ public class Frontend {
}
/**
- * Opens a Kudu transaction.
+ * Opens or continue a Kudu transaction.
*/
- private void openKuduTransaction(TQueryCtx queryCtx, AnalysisResult analysisResult,
- FeTable targetTable, EventSequence timeline) throws TransactionException {
+ private void openOrContinueKuduTransaction(PlanCtx planCtx, TQueryCtx queryCtx,
+ AnalysisResult analysisResult, FeTable targetTable, EventSequence timeline)
+ throws TransactionException {
+
+ byte[] token = planCtx.compilationState_.getKuduTransactionToken();
+ // Continue the transaction by reusing the token.
+ if (token != null) {
+ if (analysisResult.isUpdateStmt()) {
+ analysisResult.getUpdateStmt().setKuduTransactionToken(token);
+ } else if (analysisResult.isDeleteStmt()) {
+ analysisResult.getDeleteStmt().setKuduTransactionToken(token);
+ } else {
+ analysisResult.getInsertStmt().setKuduTransactionToken(token);
+ }
+ return;
+ }
+
FeKuduTable kuduTable = (FeKuduTable) targetTable;
KuduClient client = KuduUtil.getKuduClient(kuduTable.getKuduMasterHosts());
KuduTransaction txn = null;
@@ -2246,15 +2604,19 @@ public class Frontend {
txn = client.newTransaction();
timeline.markEvent(
"Kudu transaction opened with query id: " + queryCtx.getQuery_id().toString());
+ token = txn.serialize();
if (analysisResult.isUpdateStmt()) {
- analysisResult.getUpdateStmt().setKuduTransactionToken(txn.serialize());
+ analysisResult.getUpdateStmt().setKuduTransactionToken(token);
} else if (analysisResult.isDeleteStmt()) {
- analysisResult.getDeleteStmt().setKuduTransactionToken(txn.serialize());
+ analysisResult.getDeleteStmt().setKuduTransactionToken(token);
} else {
- analysisResult.getInsertStmt().setKuduTransactionToken(txn.serialize());
+ analysisResult.getInsertStmt().setKuduTransactionToken(token);
}
kuduTxnManager_.addTransaction(queryCtx.getQuery_id(), txn);
queryCtx.setIs_kudu_transactional(true);
+
+ planCtx.compilationState_.setKuduTransactionToken(token);
+
} catch (IOException e) {
if (txn != null) txn.close();
throw new TransactionException(e.getMessage());
diff --git a/fe/src/main/java/org/apache/impala/util/ClassUtil.java b/fe/src/main/java/org/apache/impala/util/ClassUtil.java
index 5d8bcd3..dca9f4b 100644
--- a/fe/src/main/java/org/apache/impala/util/ClassUtil.java
+++ b/fe/src/main/java/org/apache/impala/util/ClassUtil.java
@@ -27,4 +27,12 @@ public class ClassUtil {
StackTraceElement stackTrace = Thread.currentThread().getStackTrace()[2];
return stackTrace.getClassName() + "." + stackTrace.getMethodName() + "()";
}
+
+ public static String getStackTraceForThread() {
+ StringBuilder stringBuilder = new StringBuilder();
+ for (StackTraceElement frame: Thread.currentThread().getStackTrace()) {
+ stringBuilder.append(frame.toString() + "\n");
+ }
+ return stringBuilder.toString();
+ }
}
diff --git a/fe/src/main/java/org/apache/impala/util/ExecutorMembershipSnapshot.java b/fe/src/main/java/org/apache/impala/util/ExecutorMembershipSnapshot.java
index dcccff9..f363c4a 100644
--- a/fe/src/main/java/org/apache/impala/util/ExecutorMembershipSnapshot.java
+++ b/fe/src/main/java/org/apache/impala/util/ExecutorMembershipSnapshot.java
@@ -24,6 +24,7 @@ import java.util.concurrent.atomic.AtomicReference;
import org.apache.impala.thrift.TExecutorGroupSet;
import org.apache.impala.thrift.TNetworkAddress;
+import org.apache.impala.thrift.TExecutorGroupSet;
import org.apache.impala.thrift.TUpdateExecutorMembershipRequest;
import org.apache.kudu.shaded.com.google.common.base.Preconditions;
@@ -67,7 +68,6 @@ public class ExecutorMembershipSnapshot {
// group size. When not using executor groups (using 'default' excutor group) or
// when 'expected_executor_group_sets' startup flag is not specified, this will
// contain a single entry with an empty group name prefix.
- // TODO: IMPALA-10992: use this info to plan queries for multiple executor group sets.
private final List<TExecutorGroupSet> exec_group_sets_;
// Used only to construct the initial ExecutorMembershipSnapshot.
@@ -116,4 +116,8 @@ public class ExecutorMembershipSnapshot {
// Return the current singleton snapshot instance.
public static ExecutorMembershipSnapshot getCluster() { return cluster_.get(); }
+
+ public static List<TExecutorGroupSet> getAllExecutorGroupSets() {
+ return cluster_.get().exec_group_sets_;
+ }
}
diff --git a/fe/src/test/java/org/apache/impala/planner/ClusterSizeTest.java b/fe/src/test/java/org/apache/impala/planner/ClusterSizeTest.java
index fc53eff..50d66a5 100644
--- a/fe/src/test/java/org/apache/impala/planner/ClusterSizeTest.java
+++ b/fe/src/test/java/org/apache/impala/planner/ClusterSizeTest.java
@@ -19,9 +19,11 @@ package org.apache.impala.planner;
import org.apache.impala.common.FrontendTestBase;
import org.apache.impala.common.ImpalaException;
+import org.apache.impala.common.RuntimeEnv;
import org.apache.impala.testutil.TestUtils;
import org.apache.impala.thrift.TExecutorGroupSet;
import org.apache.impala.thrift.TQueryCtx;
+import org.apache.impala.thrift.TQueryOptions;
import org.apache.impala.thrift.TUpdateExecutorMembershipRequest;
import org.apache.impala.util.ExecutorMembershipSnapshot;
@@ -47,6 +49,10 @@ public class ClusterSizeTest extends FrontendTestBase {
TQueryCtx queryCtx = TestUtils.createQueryContext(
"functional", System.getProperty("user.name"));
queryCtx.client_request.setStmt(stmt);
+ TQueryOptions queryOptions = queryCtx.client_request.getQuery_options();
+ // Disable the default treatment of a default group as a two-group testing
+ // environment.
+ queryOptions.setTest_replan(false);
ret = frontend_.getExplainString(queryCtx);
} catch (ImpalaException e) {
fail(e.getMessage());
diff --git a/fe/src/test/resources/fair-scheduler-2-groups.xml b/fe/src/test/resources/fair-scheduler-2-groups.xml
new file mode 100644
index 0000000..0472a31
--- /dev/null
+++ b/fe/src/test/resources/fair-scheduler-2-groups.xml
@@ -0,0 +1,13 @@
+<?xml version="1.0"?>
+<allocations>
+ <queue name="root">
+ <queue name="small">
+ </queue>
+ <queue name="large">
+ </queue>
+ </queue>
+ <queuePlacementPolicy>
+ <rule name="specified" create="false"/>
+ <rule name="reject" />
+ </queuePlacementPolicy>
+</allocations>
diff --git a/fe/src/test/resources/llama-site-2-groups.xml b/fe/src/test/resources/llama-site-2-groups.xml
new file mode 100644
index 0000000..1a48a84
--- /dev/null
+++ b/fe/src/test/resources/llama-site-2-groups.xml
@@ -0,0 +1,36 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<configuration>
+ <!-- Default values -->
+ <property>
+ <name>llama.am.throttling.maximum.placed.reservations.root.small</name>
+ <value>10</value>
+ </property>
+ <property>
+ <name>llama.am.throttling.maximum.queued.reservations.root.small</name>
+ <value>30</value>
+ </property>
+ <property>
+ <name>impala.admission-control.pool-queue-timeout-ms.root.small</name>
+ <value>100000</value>
+ </property>
+ <property>
+ <name>impala.admission-control.max-query-mem-limit.root.small</name>
+ <!-- 64 MB -->
+ <value>67108864</value>
+ </property>
+ <property>
+ <name>impala.admission-control.min-query-mem-limit.root.small</name>
+ <!-- 0MB -->
+ <value>0</value>
+ </property>
+ <property>
+ <name>impala.admission-control.max-query-mem-limit.root.large</name>
+ <!-- 8 PB -->
+ <value>9007199254740992</value>
+ </property>
+ <property>
+ <name>impala.admission-control.min-query-mem-limit.root.large</name>
+ <!-- 64MB+1 -->
+ <value>67108865</value>
+ </property>
+</configuration>
diff --git a/tests/common/skip.py b/tests/common/skip.py
index a9099e4..f174689 100644
--- a/tests/common/skip.py
+++ b/tests/common/skip.py
@@ -255,6 +255,7 @@ class SkipIfDockerizedCluster:
IS_DOCKERIZED_TEST_CLUSTER, reason="Daemon would need to access host filesystem.")
insert_acls = pytest.mark.skipif(IS_DOCKERIZED_TEST_CLUSTER,
reason="IMPALA-8384: insert ACL tests are broken on dockerised minicluster.")
+ jira = partial(pytest.mark.skipif, IS_DOCKERIZED_TEST_CLUSTER)
class SkipIfHive3:
diff --git a/tests/common/test_dimensions.py b/tests/common/test_dimensions.py
index 4cc30aa..4ac912b 100644
--- a/tests/common/test_dimensions.py
+++ b/tests/common/test_dimensions.py
@@ -197,7 +197,8 @@ def create_exec_option_dimension(cluster_sizes=ALL_CLUSTER_SIZES,
'batch_size': batch_sizes,
'disable_codegen': disable_codegen_options,
'disable_codegen_rows_threshold': disable_codegen_rows_threshold_options,
- 'num_nodes': cluster_sizes}
+ 'num_nodes': cluster_sizes,
+ 'test_replan': [1]}
if sync_ddl is not None:
exec_option_dimensions['sync_ddl'] = sync_ddl
diff --git a/tests/custom_cluster/test_admission_controller.py b/tests/custom_cluster/test_admission_controller.py
index 3330b3d..37da089 100644
--- a/tests/custom_cluster/test_admission_controller.py
+++ b/tests/custom_cluster/test_admission_controller.py
@@ -671,6 +671,8 @@ class TestAdmissionController(TestAdmissionControllerBase, HS2TestSuite):
# Remove num_nodes from the options to allow test case runner to set it in one of
# the test cases.
del exec_options['num_nodes']
+ # Do not turn the default cluster into 2-group one
+ exec_options['test_replan'] = 0
exec_options['num_scanner_threads'] = 1 # To make estimates consistently reproducible
self.run_test_case('QueryTest/dedicated-coord-mem-estimates', vector_copy,
unique_database)
diff --git a/tests/custom_cluster/test_coordinators.py b/tests/custom_cluster/test_coordinators.py
index cc59fa0..c6257ed 100644
--- a/tests/custom_cluster/test_coordinators.py
+++ b/tests/custom_cluster/test_coordinators.py
@@ -275,6 +275,7 @@ class TestCoordinators(CustomClusterTestSuite):
self.client = client
client.execute("SET EXPLAIN_LEVEL=2")
+ client.execute("SET TEST_REPLAN=0")
# Ensure that the plan generated always uses only the executor nodes for scanning
# Multi-partition table
diff --git a/tests/custom_cluster/test_executor_groups.py b/tests/custom_cluster/test_executor_groups.py
index a35ca8e..918ba82 100644
--- a/tests/custom_cluster/test_executor_groups.py
+++ b/tests/custom_cluster/test_executor_groups.py
@@ -681,6 +681,62 @@ class TestExecutorGroups(CustomClusterTestSuite):
second_coord_client.close()
@pytest.mark.execute_serially
+ def test_query_assignment_with_two_exec_groups(self):
+ """This test verifies that query assignment works with two executor groups with
+ different number of executors and memory limit in each."""
+ # A small query with estimated memory per host of 10MB that can run on the small
+ # executor group
+ SMALL_QUERY = "select count(*) from tpcds_parquet.date_dim;"
+ # A large query with estimated memory per host of 132MB that can only run on
+ # the large executor group.
+ LARGE_QUERY = "select * from tpcds_parquet.store_sales where ss_item_sk = 1 limit 50;"
+ # The path to resources directory which contains the admission control config files.
+ RESOURCES_DIR = os.path.join(os.environ['IMPALA_HOME'], "fe", "src", "test",
+ "resources")
+ # Define two group sets: small and large
+ fs_allocation_path = os.path.join(RESOURCES_DIR, "fair-scheduler-2-groups.xml")
+ # Define the min-query-mem-limit and max-query-mem-limit properties of the two sets:
+ # small: [0, 64MB]
+ # large: [64MB+1Byte, 8PB]
+ llama_site_path = os.path.join(RESOURCES_DIR, "llama-site-2-groups.xml")
+ # Start with a regular admission config with multiple pools and no resource limits.
+ self._restart_coordinators(num_coordinators=1,
+ extra_args="-vmodule admission-controller=3 "
+ "-expected_executor_group_sets=root.small:2,root.large:3 "
+ "-fair_scheduler_allocation_path %s "
+ "-llama_site_path %s" % (
+ fs_allocation_path, llama_site_path))
+
+ # Create fresh client
+ self.create_impala_clients()
+ # Add an exec group with a single admission slot and 2 executors.
+ self._add_executor_group("group", 2, admission_control_slots=1,
+ resource_pool="root.small", extra_args="-mem_limit=2g")
+ # Add another exec group with a single admission slot and 3 executors.
+ self._add_executor_group("group", 3, admission_control_slots=1,
+ resource_pool="root.large", extra_args="-mem_limit=2g")
+ assert self._get_num_executor_groups(only_healthy=True) == 2
+ assert self._get_num_executor_groups(only_healthy=True,
+ exec_group_set_prefix="root.small") == 1
+ assert self._get_num_executor_groups(only_healthy=True,
+ exec_group_set_prefix="root.large") == 1
+
+ # Expect to run the small query on the small group
+ result = self.execute_query_expect_success(self.client, SMALL_QUERY)
+ assert "Executor Group: root.small-group" in str(result.runtime_profile)
+
+ # Expect to run the large query on the large group
+ result = self.execute_query_expect_success(self.client, LARGE_QUERY)
+ assert "Executor Group: root.large-group" in str(result.runtime_profile)
+
+ # Force to run the large query on the small group should fail
+ self.client.set_configuration({'request_pool': 'small'})
+ result = self.execute_query_expect_failure(self.client, LARGE_QUERY)
+ assert "The query does not fit any executor group set" in str(result)
+
+ self.client.close()
+
+ @pytest.mark.execute_serially
def test_per_exec_group_set_metrics(self):
"""This test verifies that the metrics for each exec group set are updated
appropriately."""
diff --git a/tests/query_test/test_observability.py b/tests/query_test/test_observability.py
index 79f15a1..776e1c6 100644
--- a/tests/query_test/test_observability.py
+++ b/tests/query_test/test_observability.py
@@ -356,6 +356,8 @@ class TestObservability(ImpalaTestSuite):
r'Distributed plan created:',
r'Lineage info computed:',
r'Planning finished:']
+ # Disable auto-scaling as it can produce multiple event sequences in event_regexes
+ self.execute_query_expect_success(self.client, "set enable_replan=0")
query = "select * from functional.alltypes"
runtime_profile = self.execute_query(query).runtime_profile
self.__verify_profile_event_sequence(event_regexes, runtime_profile)
@@ -765,6 +767,18 @@ class TestObservability(ImpalaTestSuite):
probe_rows_at_hdfs_scan = 'HDFS_SCAN_NODE.*\n.*Skew details: RowsRead'
assert len(re.findall(probe_rows_at_hdfs_scan, results.runtime_profile, re.M)) == 1
+ def test_query_profile_contains_auto_scaling_events(self):
+ """Test that the info about two compilation events appears in the profile."""
+ query = "select * from functional.alltypes"
+ # Specifically turn on test_replan to impose a 2-executor group environment in FE.
+ query_opts = {'enable_replan': 1, 'test_replan': 1}
+ results = self.execute_query(query, query_opts)
+ assert results.success
+ runtime_profile = results.runtime_profile
+ assert len(re.findall('Analysis finished:', runtime_profile, re.M)) == 2
+ assert len(re.findall('Single node plan created:', runtime_profile, re.M)) == 2
+ assert len(re.findall('Distributed plan created:', runtime_profile, re.M)) == 2
+
class TestQueryStates(ImpalaTestSuite):
"""Test that the 'Query State' and 'Impala Query State' are set correctly in the
runtime profile."""
diff --git a/tests/stress/test_acid_stress.py b/tests/stress/test_acid_stress.py
index f6439ff..3c3ce3a 100644
--- a/tests/stress/test_acid_stress.py
+++ b/tests/stress/test_acid_stress.py
@@ -23,7 +23,8 @@ from multiprocessing import Value
from tests.common.impala_test_suite import ImpalaTestSuite
from tests.common.parametrize import UniqueDatabase
-from tests.common.skip import SkipIf, SkipIfHive2, SkipIfS3, SkipIfGCS, SkipIfCOS
+from tests.common.skip import (SkipIf, SkipIfHive2, SkipIfS3, SkipIfGCS, SkipIfCOS,
+ SkipIfDockerizedCluster)
from tests.stress.stress_util import Task, run_tasks
NUM_OVERWRITES = 2
@@ -281,6 +282,7 @@ class TestConcurrentAcidInserts(TestAcidStress):
@SkipIfGCS.jira(reason="IMPALA-10563")
@SkipIfCOS.jira(reason="IMPALA-10773")
@SkipIfHive2.acid
+ @SkipIfDockerizedCluster.jira(reason="IMPALA-11189")
@pytest.mark.execute_serially
@pytest.mark.stress
@UniqueDatabase.parametrize(sync_ddl=True)
@@ -377,6 +379,7 @@ class TestFailingAcidInserts(TestAcidStress):
@SkipIfGCS.jira(reason="IMPALA-10563")
@SkipIfCOS.jira(reason="IMPALA-10773")
+ @SkipIfDockerizedCluster.jira(reason="IMPALA-11191")
@SkipIfHive2.acid
@pytest.mark.execute_serially
@pytest.mark.stress
[impala] 03/03: IMPALA-11192: Batch uploading files in test_scanner_fuzz.py
Posted by st...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
stigahuang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 7273c9f8a9698609816b680a79f21f9389bb265f
Author: stiga-huang <hu...@gmail.com>
AuthorDate: Thu Mar 17 10:44:58 2022 +0800
IMPALA-11192: Batch uploading files in test_scanner_fuzz.py
test_scanner_fuzz.py runs much slower on ORC than other formats. The
majority of the time is spent in uploading local files one by one to the
hdfs table folder.
The local files are copied from hdfs and randomly corrupted by the test.
The directory layout remains the same as the table folder. There are no
staging dirs that we should skip. So we can upload the whole local
folder at once, which saves a lot of the test time.
Tested locally and verified profiles of the succeeded queries. They all
scan the expected number of rows.
Change-Id: I504e160b84b3cc01d3be0b4e242d3c372692d181
Reviewed-on: http://gerrit.cloudera.org:8080/18329
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
tests/query_test/test_scanners_fuzz.py | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/tests/query_test/test_scanners_fuzz.py b/tests/query_test/test_scanners_fuzz.py
index 9c4b48a..0576132 100644
--- a/tests/query_test/test_scanners_fuzz.py
+++ b/tests/query_test/test_scanners_fuzz.py
@@ -191,7 +191,7 @@ class TestScannersFuzzing(ImpalaTestSuite):
table_loc = self._get_table_location(fq_fuzz_table_name, vector)
check_call(['hdfs', 'dfs', '-copyToLocal', table_loc + "/*", tmp_table_dir])
partitions = self.walk_and_corrupt_table_data(tmp_table_dir, num_copies, rng)
- self.path_aware_copy_files_to_hdfs(tmp_table_dir, table_loc)
+ self.filesystem_client.copy_from_local(tmp_table_dir, table_loc)
else:
self.execute_query("create table %s.%s like %s.%s" % (fuzz_db, fuzz_table,
src_db, src_table))