You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by mi...@apache.org on 2024/10/15 20:11:25 UTC
(impala) branch master updated (2535e7949 -> 32cec3fd8)
This is an automated email from the ASF dual-hosted git repository.
michaelsmith pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
from 2535e7949 IMPALA-12216: Print timestamp for impala-shell errors
new 40bb93fc4 IMPALA-12146: Fix incorrect host memory reserved when the executor quits abnormally
new c5b7c6395 IMPALA-11943: Mark utf8 string functions with IR_ALWAYS_INLINE
new 32cec3fd8 IMPALA-13429: Calcite planner crashing with outer join
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/exprs/mask-functions-ir.cc | 75 ++++++++++++--------
be/src/exprs/string-functions-ir.cc | 17 +++--
be/src/scheduling/admission-controller-test.cc | 81 ++++++++++++++++++++++
be/src/scheduling/admission-controller.cc | 31 +++++++--
be/src/scheduling/admission-controller.h | 11 ++-
.../impala/calcite/rel/node/ImpalaJoinRel.java | 59 ++++++++++++++--
.../calcite/rel/util/RexInputRefCollector.java | 38 ++++++----
.../workloads/targeted-perf/queries/string.test | 60 ++++++++++++++++
8 files changed, 312 insertions(+), 60 deletions(-)
copy fe/src/main/java/org/apache/impala/catalog/Role.java => java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/util/RexInputRefCollector.java (50%)
(impala) 03/03: IMPALA-13429: Calcite planner crashing with outer join
Posted by mi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
michaelsmith pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 32cec3fd80b129e4d84a403f1ee0c4144101f80c
Author: Steve Carlin <sc...@cloudera.com>
AuthorDate: Tue Oct 8 13:59:41 2024 -0700
IMPALA-13429: Calcite planner crashing with outer join
The Calcite planner was crashing when there was an outer join
and there was a conjunct that compared two columns within the
same table. This conjunct needs to be place in "other" conjuncts
rather than "equi" conjuncts.
Change-Id: I4ae2d257fa58f3a58079b6aa551c32ffda7d28cf
Reviewed-on: http://gerrit.cloudera.org:8080/21908
Reviewed-by: Michael Smith <mi...@cloudera.com>
Tested-by: Michael Smith <mi...@cloudera.com>
---
.../impala/calcite/rel/node/ImpalaJoinRel.java | 59 +++++++++++++++++++---
.../calcite/rel/util/RexInputRefCollector.java | 51 +++++++++++++++++++
2 files changed, 104 insertions(+), 6 deletions(-)
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/ImpalaJoinRel.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/ImpalaJoinRel.java
index bcaf7241b..4903c64a0 100644
--- a/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/ImpalaJoinRel.java
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/ImpalaJoinRel.java
@@ -46,6 +46,7 @@ import org.apache.impala.calcite.rel.phys.ImpalaHashJoinNode;
import org.apache.impala.calcite.rel.phys.ImpalaNestedLoopJoinNode;
import org.apache.impala.calcite.rel.util.CreateExprVisitor;
import org.apache.impala.calcite.rel.util.ExprConjunctsConverter;
+import org.apache.impala.calcite.rel.util.RexInputRefCollector;
import org.apache.impala.calcite.type.ImpalaTypeConverter;
import org.apache.impala.catalog.Function;
import org.apache.impala.catalog.Type;
@@ -57,6 +58,7 @@ import org.apache.impala.planner.PlanNode;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
+import java.util.Set;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -98,7 +100,7 @@ public class ImpalaJoinRel extends Join
JoinOperator joinOp = getImpalaJoinOp();
List<BinaryPredicate> equiJoinConjuncts = new ArrayList<>();
- List<Expr> nonEquiJoinConjuncts = new ArrayList<>();
+ List<Expr> otherJoinConjuncts = new ArrayList<>();
// IMPALA-13176: TODO: Impala allows forcing hints for the distribution mode
// - e.g force broadcast or hash partition join. However, we are not
@@ -119,7 +121,7 @@ public class ImpalaJoinRel extends Join
if (conjunctInfo.isEquiJoin_) {
equiJoinConjuncts.add((BinaryPredicate) conjunctInfo.conjunct_);
} else {
- nonEquiJoinConjuncts.add(conjunctInfo.conjunct_);
+ otherJoinConjuncts.add(conjunctInfo.conjunct_);
}
}
}
@@ -136,10 +138,10 @@ public class ImpalaJoinRel extends Join
PlanNode joinNode = equiJoinConjuncts.size() == 0
? new ImpalaNestedLoopJoinNode(context.ctx_.getNextNodeId(), leftInput.planNode_,
rightInput.planNode_, false /* not a straight join */, distMode, joinOp,
- nonEquiJoinConjuncts, filterConjuncts, analyzer)
+ otherJoinConjuncts, filterConjuncts, analyzer)
: new ImpalaHashJoinNode(context.ctx_.getNextNodeId(), leftInput.planNode_,
rightInput.planNode_, false /* not a straight join */, distMode, joinOp,
- equiJoinConjuncts, nonEquiJoinConjuncts, filterConjuncts, analyzer);
+ equiJoinConjuncts, otherJoinConjuncts, filterConjuncts, analyzer);
if (equiJoinConjuncts.size() > 0) {
// register the equi and non-equi join conjuncts with the analyzer such that
@@ -147,7 +149,7 @@ public class ImpalaJoinRel extends Join
List<Expr> equiJoinExprs = new ArrayList<Expr>(equiJoinConjuncts);
analyzer.registerConjuncts(getJoinConjunctListToRegister(equiJoinExprs));
}
- analyzer.registerConjuncts(getJoinConjunctListToRegister(nonEquiJoinConjuncts));
+ analyzer.registerConjuncts(getJoinConjunctListToRegister(otherJoinConjuncts));
return new NodeWithExprs(joinNode, outputExprs);
}
@@ -425,11 +427,56 @@ public class ImpalaJoinRel extends Join
// get a canonicalized representation
conj = getCanonical(conj, leftInput.outputExprs_.size());
Expr impalaConjunct = conj.accept(visitor);
- conjunctInfos.add(new ConjunctInfo(impalaConjunct, conj.isA(SqlKind.EQUALS)));
+ conjunctInfos.add(
+ new ConjunctInfo(impalaConjunct, isEquijoinConjunct(conj, leftInput)));
}
return conjunctInfos;
}
+ // Checks to see if this conjunct can be considered an "equijoin" conjunct.
+ // An equijoin conjunct must have the following traits:
+ // - should be an "EQUALS" RexCall at the top level.
+ // - be in canonical form, that is, the input refs on the left side of the
+ // equals RexNode should point to the left input and the right side of the
+ // equals RexNode should point to the right input.
+ // - should be at least one input ref on the left side and one input ref on
+ // the right side.
+ private boolean isEquijoinConjunct(RexNode conjunct, NodeWithExprs leftInput) {
+ if (!conjunct.isA(SqlKind.EQUALS)) {
+ return false;
+ }
+
+ Preconditions.checkState(conjunct instanceof RexCall);
+ RexCall call = (RexCall) conjunct;
+
+ // left side expr may only contain inputrefs from left side.
+ // right side expr may only contain inputrefs from right side.
+ RexNode left = call.getOperands().get(0);
+ Set<Integer> inputRefs = RexInputRefCollector.getInputRefs(left);
+ if (inputRefs.size() == 0) {
+ return false;
+ }
+
+ for (Integer inputRef : inputRefs) {
+ if (inputRef >= leftInput.outputExprs_.size()) {
+ return false;
+ }
+ }
+
+ RexNode right = call.getOperands().get(1);
+ inputRefs = RexInputRefCollector.getInputRefs(right);
+ if (inputRefs.size() == 0) {
+ return false;
+ }
+ for (Integer inputRef : inputRefs) {
+ if (inputRef < leftInput.outputExprs_.size()) {
+ return false;
+ }
+ }
+ return true;
+
+ }
+
private static class ConjunctInfo {
public final Expr conjunct_;
public final boolean isEquiJoin_;
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/util/RexInputRefCollector.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/util/RexInputRefCollector.java
new file mode 100644
index 000000000..f930cb435
--- /dev/null
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/util/RexInputRefCollector.java
@@ -0,0 +1,51 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.impala.calcite.rel.util;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexVisitorImpl;
+
+/**
+ * Collects all inputrefs for a given RexNodeExpr through the only
+ * public method, RexInputRefCollector.getInputRefs(RexNode).
+ */
+public class RexInputRefCollector extends RexVisitorImpl<Void> {
+
+ private final Set<Integer> inputRefSet = new HashSet<Integer>();
+
+ private RexInputRefCollector(boolean deep) {
+ super(deep);
+ }
+
+ @Override
+ public Void visitInputRef(RexInputRef inputRef) {
+ inputRefSet.add(inputRef.getIndex());
+ return null;
+ }
+
+ public static Set<Integer> getInputRefs(RexNode expr) {
+ RexInputRefCollector collector = new RexInputRefCollector(true);
+ expr.accept(collector);
+ return collector.inputRefSet;
+ }
+
+}
(impala) 02/03: IMPALA-11943: Mark utf8 string functions with IR_ALWAYS_INLINE
Posted by mi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
michaelsmith pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit c5b7c6395b1f161bbc55596bb3694fbcf3e658c6
Author: stiga-huang <hu...@gmail.com>
AuthorDate: Fri Feb 24 15:59:59 2023 +0800
IMPALA-11943: Mark utf8 string functions with IR_ALWAYS_INLINE
String functions that have both UTF-8 and the traditional ASCII
behaviors have checks for the UTF8_MODE query option. The check is
intended to be replaced with constants during codegen in
LlvmCodeGen::InlineConstFnAttrs().
However, as mentioned in the method comment, InlineConstFnAttrs() only
replaces call instructions inside the current function. To replace the
call to FunctionContextImpl::GetConstFnAttr() inside the callee
functions, we have to inline the callee functions (by annotating them
with IR_ALWAYS_INLINE).
This patch annotates UTF-8 related string functions with
IR_ALWAYS_INLINE to make sure the checks on UTF8_MODE are all replaced
in codegen. Note that builtin functions don't need the annotation if
they are not invoked by other builtin functions, because
InlineConstFnAttrs() will be invoked recursively in the expression tree.
See ScalarFnCall::GetCodegendComputeFnImpl().
Perf tests:
Ran PERF_STRING queries in targeted-perf (scale factor 100) on parquet
format. Saw significant improvements:
+-----------------+--------------+--------+-------------+------------+-----------+----------------+-------+----------------+---------+---------+
| Query | File Format | Avg(s) | Base Avg(s) | Delta(Avg) | StdDev(%) | Base StdDev(%) | Iters | Median Diff(%) | MW Zval | Tval |
+-----------------+--------------+--------+-------------+------------+-----------+----------------+-------+----------------+---------+---------+
| PERF_STRING-Q6 | parquet/none | 11.98 | 12.53 | -4.39% | 0.44% | 0.51% | 30 | -4.59% | -6.54 | -36.46 |
| PERF_STRING-Q9 | parquet/none | 11.76 | 12.35 | -4.77% | 0.38% | 0.41% | 30 | -5.04% | -6.54 | -47.82 |
| PERF_STRING-Q7 | parquet/none | 9.88 | 10.44 | I -5.34% | 0.83% | 1.10% | 30 | I -5.64% | -6.54 | -21.64 |
| PERF_STRING-Q13 | parquet/none | 9.52 | 10.08 | I -5.56% | 0.55% | 0.59% | 30 | I -5.89% | -6.54 | -38.72 |
| PERF_STRING-Q11 | parquet/none | 10.97 | 11.64 | I -5.72% | 0.44% | 0.61% | 30 | I -6.00% | -6.54 | -42.72 |
| PERF_STRING-Q4 | parquet/none | 5.30 | 5.66 | I -6.33% | 1.06% | 1.44% | 30 | I -6.49% | -6.54 | -19.84 |
| PERF_STRING-Q3 | parquet/none | 5.27 | 5.70 | I -7.43% | 1.03% | 0.94% | 30 | I -8.12% | -6.54 | -30.40 |
| PERF_STRING-Q5 | parquet/none | 5.87 | 6.36 | I -7.69% | 1.18% | 0.88% | 30 | I -8.47% | -6.54 | -30.09 |
| PERF_STRING-Q12 | parquet/none | 6.56 | 7.15 | I -8.33% | 0.67% | 0.73% | 30 | I -9.03% | -6.54 | -47.94 |
| PERF_STRING-Q10 | parquet/none | 6.62 | 7.24 | I -8.54% | 0.84% | 0.60% | 30 | I -9.34% | -6.54 | -48.21 |
| PERF_STRING-Q2 | parquet/none | 4.77 | 5.25 | I -9.20% | 0.89% | 1.05% | 30 | I -10.17% | -6.54 | -37.90 |
| PERF_STRING-Q1 | parquet/none | 4.16 | 4.62 | I -9.80% | 2.10% | 1.10% | 30 | I -11.27% | -6.54 | -24.50 |
| PERF_STRING-Q8 | parquet/none | 5.10 | 6.57 | I -22.40% | 0.87% | 0.73% | 30 | I -28.74% | -6.54 | -123.34 |
+-----------------+--------------+--------+-------------+------------+-----------+----------------+-------+----------------+---------+---------+
Note that Q8-Q13 are new queries added by this patch to reveal the
performance difference.
The perf test command is
bin/single_node_perf_run.py --iterations 30 --scale 100 \
--table_formats parquet/none --num_impalads 3 \
--query_names 'PERF_STRING.*' f4a321cf8 c8d2b6a90
Running on this branch:
https://github.com/stiga-huang/impala/commits/inline-utf8-func-perf-test
Change-Id: I19e8fba332ae329da8b1d37dba3bbc64f59e6f3a
Reviewed-on: http://gerrit.cloudera.org:8080/19535
Reviewed-by: Daniel Becker <da...@cloudera.com>
Reviewed-by: Csaba Ringhofer <cs...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/exprs/mask-functions-ir.cc | 75 +++++++++++++---------
be/src/exprs/string-functions-ir.cc | 17 +++--
.../workloads/targeted-perf/queries/string.test | 60 +++++++++++++++++
3 files changed, 117 insertions(+), 35 deletions(-)
diff --git a/be/src/exprs/mask-functions-ir.cc b/be/src/exprs/mask-functions-ir.cc
index b6725ea5c..689da79d0 100644
--- a/be/src/exprs/mask-functions-ir.cc
+++ b/be/src/exprs/mask-functions-ir.cc
@@ -23,6 +23,7 @@
#include <openssl/err.h>
#include <openssl/sha.h>
+#include "codegen/impala-ir.h"
#include "exprs/anyval-util.h"
#include "exprs/math-functions.h"
#include "exprs/string-functions.h"
@@ -213,9 +214,9 @@ static int GetUtf8CodePointCount(FunctionContext* ctx, const StringVal& val) {
/// Mask the given string except the first 'un_mask_char_count' chars. Ported from
/// org.apache.hadoop.hive.ql.udf.generic.GenericUDFMaskShowFirstN.
-static inline StringVal MaskShowFirstNImpl(FunctionContext* ctx, const StringVal& val,
- int un_mask_char_count, int masked_upper_char, int masked_lower_char,
- int masked_digit_char, int masked_other_char) {
+IR_ALWAYS_INLINE static inline StringVal MaskShowFirstNImpl(FunctionContext* ctx,
+ const StringVal& val, int un_mask_char_count, int masked_upper_char,
+ int masked_lower_char, int masked_digit_char, int masked_other_char) {
// To be consistent with Hive, negative char_count is treated as 0.
if (un_mask_char_count < 0) un_mask_char_count = 0;
if (val.is_null || val.len == 0 || un_mask_char_count >= val.len) return val;
@@ -229,9 +230,9 @@ static inline StringVal MaskShowFirstNImpl(FunctionContext* ctx, const StringVal
/// Mask the given string except the last 'un_mask_char_count' chars. Ported from
/// org.apache.hadoop.hive.ql.udf.generic.GenericUDFMaskShowLastN.
-static inline StringVal MaskShowLastNImpl(FunctionContext* ctx, const StringVal& val,
- int un_mask_char_count, int masked_upper_char, int masked_lower_char,
- int masked_digit_char, int masked_other_char) {
+IR_ALWAYS_INLINE static inline StringVal MaskShowLastNImpl(FunctionContext* ctx,
+ const StringVal& val, int un_mask_char_count, int masked_upper_char,
+ int masked_lower_char, int masked_digit_char, int masked_other_char) {
// To be consistent with Hive, negative char_count is treated as 0.
if (un_mask_char_count < 0) un_mask_char_count = 0;
if (val.is_null || val.len == 0 || un_mask_char_count >= val.len) return val;
@@ -247,9 +248,9 @@ static inline StringVal MaskShowLastNImpl(FunctionContext* ctx, const StringVal&
/// Mask the first 'mask_char_count' chars of the given string. Ported from
/// org.apache.hadoop.hive.ql.udf.generic.GenericUDFMaskFirstN.
-static inline StringVal MaskFirstNImpl(FunctionContext* ctx, const StringVal& val,
- int mask_char_count, int masked_upper_char, int masked_lower_char,
- int masked_digit_char, int masked_other_char) {
+IR_ALWAYS_INLINE static inline StringVal MaskFirstNImpl(FunctionContext* ctx,
+ const StringVal& val, int mask_char_count, int masked_upper_char,
+ int masked_lower_char, int masked_digit_char, int masked_other_char) {
if (mask_char_count <= 0 || val.is_null || val.len == 0) return val;
if (mask_char_count > val.len) mask_char_count = val.len;
if (!ctx->impl()->GetConstFnAttr(FunctionContextImpl::UTF8_MODE)) {
@@ -262,9 +263,9 @@ static inline StringVal MaskFirstNImpl(FunctionContext* ctx, const StringVal& va
/// Mask the last 'mask_char_count' chars of the given string. Ported from
/// org.apache.hadoop.hive.ql.udf.generic.GenericUDFMaskLastN.
-static inline StringVal MaskLastNImpl(FunctionContext* ctx, const StringVal& val,
- int mask_char_count, int masked_upper_char, int masked_lower_char,
- int masked_digit_char, int masked_other_char) {
+IR_ALWAYS_INLINE static inline StringVal MaskLastNImpl(FunctionContext* ctx,
+ const StringVal& val, int mask_char_count, int masked_upper_char,
+ int masked_lower_char, int masked_digit_char, int masked_other_char) {
if (mask_char_count <= 0 || val.is_null || val.len == 0) return val;
if (mask_char_count > val.len) mask_char_count = val.len;
if (!ctx->impl()->GetConstFnAttr(FunctionContextImpl::UTF8_MODE)) {
@@ -279,9 +280,9 @@ static inline StringVal MaskLastNImpl(FunctionContext* ctx, const StringVal& val
/// Mask the whole given string. Ported from
/// org.apache.hadoop.hive.ql.udf.generic.GenericUDFMask.
-static inline StringVal MaskImpl(FunctionContext* ctx, const StringVal& val,
- int masked_upper_char, int masked_lower_char, int masked_digit_char,
- int masked_other_char) {
+IR_ALWAYS_INLINE static inline StringVal MaskImpl(FunctionContext* ctx,
+ const StringVal& val, int masked_upper_char, int masked_lower_char,
+ int masked_digit_char, int masked_other_char) {
if (val.is_null || val.len == 0) return val;
if (!ctx->impl()->GetConstFnAttr(FunctionContextImpl::UTF8_MODE)) {
return MaskSubStr(ctx, val, 0, val.len, masked_upper_char,
@@ -410,8 +411,8 @@ static DateVal MaskImpl(FunctionContext* ctx, const DateVal& val, int day_value,
/// Gets the first character of 'str'. Returns 'default_value' if 'str' is empty.
/// In UTF-8 mode, the first code point is returned.
/// Otherwise, the first char is returned.
-static inline uint32_t GetFirstChar(FunctionContext* ctx, const StringVal& str,
- uint32_t default_value) {
+IR_ALWAYS_INLINE static inline uint32_t GetFirstChar(FunctionContext* ctx,
+ const StringVal& str, uint32_t default_value) {
// To be consistent with Hive, empty string is converted to default value. String with
// length > 1 will only use its first char.
if (str.len == 0) return default_value;
@@ -453,9 +454,12 @@ StringVal MaskFunctions::MaskShowFirstN(FunctionContext* ctx, const StringVal& v
return MaskShowFirstNImpl(ctx, val, un_mask_char_count, MASKED_UPPERCASE,
MASKED_LOWERCASE, MASKED_DIGIT, MASKED_OTHER_CHAR);
}
-StringVal MaskFunctions::MaskShowFirstN(FunctionContext* ctx, const StringVal& val,
- const IntVal& char_count, const StringVal& upper_char, const StringVal& lower_char,
- const StringVal& digit_char, const StringVal& other_char) {
+// Marked as IR_ALWAYS_INLINE since this is called in another overload.
+// This way the overload can also replace GetConstFnAttr() calls in codegen.
+IR_ALWAYS_INLINE StringVal MaskFunctions::MaskShowFirstN(FunctionContext* ctx,
+ const StringVal& val, const IntVal& char_count, const StringVal& upper_char,
+ const StringVal& lower_char, const StringVal& digit_char,
+ const StringVal& other_char) {
return MaskShowFirstNImpl(ctx, val, char_count.val,
GetFirstChar(ctx, upper_char, MASKED_UPPERCASE),
GetFirstChar(ctx, lower_char, MASKED_LOWERCASE),
@@ -540,9 +544,12 @@ StringVal MaskFunctions::MaskShowLastN(FunctionContext* ctx, const StringVal& va
return MaskShowLastNImpl(ctx, val, un_mask_char_count, MASKED_UPPERCASE,
MASKED_LOWERCASE, MASKED_DIGIT, MASKED_OTHER_CHAR);
}
-StringVal MaskFunctions::MaskShowLastN(FunctionContext* ctx, const StringVal& val,
- const IntVal& char_count, const StringVal& upper_char, const StringVal& lower_char,
- const StringVal& digit_char, const StringVal& other_char) {
+// Marked as IR_ALWAYS_INLINE since this is called in another overload.
+// This way the overload can also replace GetConstFnAttr() calls in codegen.
+IR_ALWAYS_INLINE StringVal MaskFunctions::MaskShowLastN(FunctionContext* ctx,
+ const StringVal& val, const IntVal& char_count, const StringVal& upper_char,
+ const StringVal& lower_char, const StringVal& digit_char,
+ const StringVal& other_char) {
return MaskShowLastNImpl(ctx, val, char_count.val,
GetFirstChar(ctx, upper_char, MASKED_UPPERCASE),
GetFirstChar(ctx, lower_char, MASKED_LOWERCASE),
@@ -617,9 +624,12 @@ StringVal MaskFunctions::MaskFirstN(FunctionContext* ctx, const StringVal& val,
return MaskFirstNImpl(ctx, val, char_count.val, MASKED_UPPERCASE, MASKED_LOWERCASE,
MASKED_DIGIT, MASKED_OTHER_CHAR);
}
-StringVal MaskFunctions::MaskFirstN(FunctionContext* ctx, const StringVal& val,
- const IntVal& char_count, const StringVal& upper_char, const StringVal& lower_char,
- const StringVal& digit_char, const StringVal& other_char) {
+// Marked as IR_ALWAYS_INLINE since this is called in another overload.
+// This way the overload can also replace GetConstFnAttr() calls in codegen.
+IR_ALWAYS_INLINE StringVal MaskFunctions::MaskFirstN(FunctionContext* ctx,
+ const StringVal& val, const IntVal& char_count, const StringVal& upper_char,
+ const StringVal& lower_char, const StringVal& digit_char,
+ const StringVal& other_char) {
return MaskFirstNImpl(ctx, val, char_count.val,
GetFirstChar(ctx, upper_char, MASKED_UPPERCASE),
GetFirstChar(ctx, lower_char, MASKED_LOWERCASE),
@@ -694,9 +704,12 @@ StringVal MaskFunctions::MaskLastN(FunctionContext* ctx, const StringVal& val,
return MaskLastNImpl(ctx, val, char_count.val, MASKED_UPPERCASE, MASKED_LOWERCASE,
MASKED_DIGIT, MASKED_OTHER_CHAR);
}
-StringVal MaskFunctions::MaskLastN(FunctionContext* ctx, const StringVal& val,
- const IntVal& char_count, const StringVal& upper_char, const StringVal& lower_char,
- const StringVal& digit_char, const StringVal& other_char) {
+// Marked as IR_ALWAYS_INLINE since this is called in another overload.
+// This way the overload can also replace GetConstFnAttr() calls in codegen.
+IR_ALWAYS_INLINE StringVal MaskFunctions::MaskLastN(FunctionContext* ctx,
+ const StringVal& val, const IntVal& char_count, const StringVal& upper_char,
+ const StringVal& lower_char, const StringVal& digit_char,
+ const StringVal& other_char) {
return MaskLastNImpl(ctx, val, char_count.val,
GetFirstChar(ctx, upper_char, MASKED_UPPERCASE),
GetFirstChar(ctx, lower_char, MASKED_LOWERCASE),
@@ -766,7 +779,9 @@ StringVal MaskFunctions::Mask(FunctionContext* ctx, const StringVal& val) {
return MaskImpl(ctx, val, MASKED_UPPERCASE, MASKED_LOWERCASE, MASKED_DIGIT,
MASKED_OTHER_CHAR);
}
-StringVal MaskFunctions::Mask(FunctionContext* ctx, const StringVal& val,
+// Marked as IR_ALWAYS_INLINE since this is called in other overloads.
+// This way the overloads can also replace GetConstFnAttr() calls in codegen.
+IR_ALWAYS_INLINE StringVal MaskFunctions::Mask(FunctionContext* ctx, const StringVal& val,
const StringVal& upper_char, const StringVal& lower_char,
const StringVal& digit_char, const StringVal& other_char) {
return MaskImpl(ctx, val,
diff --git a/be/src/exprs/string-functions-ir.cc b/be/src/exprs/string-functions-ir.cc
index dc9468328..0a2319483 100644
--- a/be/src/exprs/string-functions-ir.cc
+++ b/be/src/exprs/string-functions-ir.cc
@@ -57,7 +57,9 @@ uint64_t StringFunctions::re2_mem_limit_ = 8 << 20;
// - 1-indexed positions
// - supported negative positions (count from the end of the string)
// - [optional] len. No len indicates longest substr possible
-StringVal StringFunctions::Substring(FunctionContext* context,
+// Marks as IR_ALWAYS_INLINE since this is called in other overloads.
+// So the overloads can also replace GetConstFnAttr() calls in codegen.
+IR_ALWAYS_INLINE StringVal StringFunctions::Substring(FunctionContext* context,
const StringVal& str, const BigIntVal& pos, const BigIntVal& len) {
if (str.is_null || pos.is_null || len.is_null) return StringVal::null();
if (context->impl()->GetConstFnAttr(FunctionContextImpl::UTF8_MODE)) {
@@ -257,7 +259,7 @@ StringVal StringFunctions::Rpad(FunctionContext* context, const StringVal& str,
IntVal StringFunctions::Length(FunctionContext* context, const StringVal& str) {
if (str.is_null) return IntVal::null();
- if (context->impl()->GetConstFnAttr(FunctionContextImpl::UTF8_MODE, 0)) {
+ if (context->impl()->GetConstFnAttr(FunctionContextImpl::UTF8_MODE)) {
return Utf8Length(context, str);
}
return IntVal(str.len);
@@ -288,7 +290,10 @@ IntVal StringFunctions::Utf8Length(FunctionContext* context, const StringVal& st
return IntVal(CountUtf8Chars(str.ptr, str.len));
}
-StringVal StringFunctions::Lower(FunctionContext* context, const StringVal& str) {
+// Marks as IR_ALWAYS_INLINE since this is called in MaskFunctions::MaskHash().
+// So the caller can also replace GetConstFnAttr() calls in codegen.
+IR_ALWAYS_INLINE StringVal StringFunctions::Lower(FunctionContext* context,
+ const StringVal& str) {
if (str.is_null) return StringVal::null();
if (context->impl()->GetConstFnAttr(FunctionContextImpl::UTF8_MODE)) {
return LowerUtf8(context, str);
@@ -879,8 +884,10 @@ IntVal StringFunctions::Ascii(FunctionContext* context, const StringVal& str) {
return IntVal((str.len == 0) ? 0 : static_cast<int32_t>(str.ptr[0]));
}
-IntVal StringFunctions::Instr(FunctionContext* context, const StringVal& str,
- const StringVal& substr, const BigIntVal& start_position,
+// Marks as IR_ALWAYS_INLINE since this is called in other overloads.
+// So the overloads can also replace GetConstFnAttr() calls in codegen.
+IR_ALWAYS_INLINE IntVal StringFunctions::Instr(FunctionContext* context,
+ const StringVal& str, const StringVal& substr, const BigIntVal& start_position,
const BigIntVal& occurrence) {
if (str.is_null || substr.is_null || start_position.is_null || occurrence.is_null) {
return IntVal::null();
diff --git a/testdata/workloads/targeted-perf/queries/string.test b/testdata/workloads/targeted-perf/queries/string.test
index c42c31aca..f6f0faf03 100644
--- a/testdata/workloads/targeted-perf/queries/string.test
+++ b/testdata/workloads/targeted-perf/queries/string.test
@@ -56,3 +56,63 @@ WHERE lower(l_comment) = 'egular courts above the'
---- TYPES
BIGINT
====
+---- QUERY: PERF_STRING-Q8
+-- Make sure utf8 string functions are inlined in jitted codes so the checks for
+-- utf8 mode are replaced with constants
+SELECT count(*) FROM lineitem
+WHERE instr(l_comment, 'egular courts above the') > 0
+---- RESULTS
+28
+---- TYPES
+BIGINT
+====
+---- QUERY: PERF_STRING-Q9
+-- Make sure utf8 string functions are inlined in jitted codes so the checks for
+-- utf8 mode are replaced with constants
+SELECT count(*) FROM lineitem
+WHERE mask(l_comment) is null
+---- RESULTS
+0
+---- TYPES
+BIGINT
+====
+---- QUERY: PERF_STRING-Q10
+-- Make sure utf8 string functions are inlined in jitted codes so the checks for
+-- utf8 mode are replaced with constants
+SELECT count(*) FROM lineitem
+WHERE mask_first_n(l_comment) is null
+---- RESULTS
+0
+---- TYPES
+BIGINT
+====
+---- QUERY: PERF_STRING-Q11
+-- Make sure utf8 string functions are inlined in jitted codes so the checks for
+-- utf8 mode are replaced with constants
+SELECT count(*) FROM lineitem
+WHERE mask_show_first_n(l_comment) is null
+---- RESULTS
+0
+---- TYPES
+BIGINT
+====
+---- QUERY: PERF_STRING-Q12
+-- Make sure utf8 string functions are inlined in jitted codes so the checks for
+-- utf8 mode are replaced with constants
+SELECT count(*) FROM lineitem
+WHERE mask_last_n(l_comment) is null
+---- RESULTS
+0
+---- TYPES
+BIGINT
+====
+---- QUERY: PERF_STRING-Q13
+-- Make sure utf8 string functions are inlined in jitted codes so the checks for
+-- utf8 mode are replaced with constants
+SELECT count(*) FROM lineitem
+WHERE mask_show_last_n(l_comment) is null
+---- RESULTS
+0
+---- TYPES
+BIGINT
+====
(impala) 01/03: IMPALA-12146: Fix incorrect host memory reserved when the executor quits abnormally
Posted by mi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
michaelsmith pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 40bb93fc4beb3b216aca942a01352feb535a6cb8
Author: Yida Wu <yi...@cloudera.com>
AuthorDate: Tue Oct 1 11:55:46 2024 -0700
IMPALA-12146: Fix incorrect host memory reserved when the executor quits abnormally
Currently there is an issue where if an executor quits abnormally
while running a query, its reserved memory may still remain in the
coordinator's host stats.
The remote aggregated memory reserved uses all available remote
pool stats for calculation. The problem happens when the statestore
sends a topic update to update the pool stats. Although the
coordinator removes the remote stats from the pool for the terminated
executor during the update, in UpdateClusterAggregates(), it fails to
reset the corresponding aggregated memory reserved for that host
if all the remote stats for that host have been removed. This can
lead to stale memory reserved value remaining.
To fix this, added a logic to ensure that the stats of memory
reserved of that host are reset in the aggregated host stats when a
delete topic for the host is detected and the host no longer exists
in any remote pool stats.
Tests:
Passed exhaustive tests.
Added testcase AdmissionControllerTest::EraseHostStats.
Manually verified that the coordinator web ui correctly showed the
reserved memory after the crashed executor recovered and rejoined.
Change-Id: Ic6f6edd28c55904d63d0c494230ee2bf7a0f6cce
Reviewed-on: http://gerrit.cloudera.org:8080/21896
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Reviewed-by: Michael Smith <mi...@cloudera.com>
---
be/src/scheduling/admission-controller-test.cc | 81 ++++++++++++++++++++++++++
be/src/scheduling/admission-controller.cc | 31 ++++++++--
be/src/scheduling/admission-controller.h | 11 +++-
3 files changed, 117 insertions(+), 6 deletions(-)
diff --git a/be/src/scheduling/admission-controller-test.cc b/be/src/scheduling/admission-controller-test.cc
index 545002f1b..0281608ed 100644
--- a/be/src/scheduling/admission-controller-test.cc
+++ b/be/src/scheduling/admission-controller-test.cc
@@ -546,6 +546,87 @@ TEST_F(AdmissionControllerTest, Simple) {
ASSERT_FALSE(coordinator_resource_limited);
}
+/// Test that removing hosts from AdmissionController, and check if the memory reserved in
+/// host stats for the specific host is updated correctly after the host is removed.
+TEST_F(AdmissionControllerTest, EraseHostStats) {
+ FLAGS_fair_scheduler_allocation_path = GetResourceFile("fair-scheduler-test2.xml");
+ FLAGS_llama_site_path = GetResourceFile("llama-site-test2.xml");
+
+ AdmissionController* admission_controller = MakeAdmissionController();
+
+ ASSERT_EQ(0, admission_controller->host_stats_.size());
+
+ TTopicDelta membership = MakeTopicDelta(false);
+
+ AddStatsToTopic(&membership, HOST_1, QUEUE_C, MakePoolStats(1000, 1, 0));
+ AddStatsToTopic(&membership, HOST_2, QUEUE_C, MakePoolStats(5000, 10, 0));
+ AddStatsToTopic(&membership, HOST_2, QUEUE_B, MakePoolStats(3000, 5, 0));
+
+ StatestoreSubscriber::TopicDeltaMap initial_topic_deltas;
+ initial_topic_deltas.emplace(Statestore::IMPALA_REQUEST_QUEUE_TOPIC, membership);
+ vector<TTopicDelta> outgoing_topic_updates;
+ admission_controller->UpdatePoolStats(initial_topic_deltas, &outgoing_topic_updates);
+
+ // Verify that the host stats were added
+ ASSERT_EQ(3, admission_controller->host_stats_.size());
+ ASSERT_EQ(1, admission_controller->host_stats_.count(HOST_0));
+ ASSERT_EQ(1, admission_controller->host_stats_.count(HOST_1));
+ ASSERT_EQ(1, admission_controller->host_stats_.count(HOST_2));
+ ASSERT_EQ(1000, admission_controller->host_stats_[HOST_1].mem_reserved);
+ ASSERT_EQ(8000, admission_controller->host_stats_[HOST_2].mem_reserved);
+
+ // Create an update that deletes HOST_1
+ TTopicDelta delete_update = MakeTopicDelta(true);
+ TTopicItem delete_item;
+ delete_item.key = "POOL:" + QUEUE_C + "!" + HOST_1;
+ delete_item.deleted = true;
+ delete_update.topic_entries.push_back(delete_item);
+
+ StatestoreSubscriber::TopicDeltaMap delete_topic_deltas;
+ delete_topic_deltas.emplace(Statestore::IMPALA_REQUEST_QUEUE_TOPIC, delete_update);
+
+ // Apply the delete update
+ admission_controller->UpdatePoolStats(delete_topic_deltas, &outgoing_topic_updates);
+
+ // Verify that HOST_1 mem_reserved was reset.
+ ASSERT_EQ(3, admission_controller->host_stats_.size());
+ ASSERT_EQ(8000, admission_controller->host_stats_[HOST_2].mem_reserved);
+ ASSERT_EQ(0, admission_controller->host_stats_[HOST_1].mem_reserved);
+
+ // Verify that the pool stats were updated accordingly
+ AdmissionController::PoolStats* pool_stats =
+ admission_controller->GetPoolStats(QUEUE_C);
+ ASSERT_EQ(5000, pool_stats->agg_mem_reserved_);
+ ASSERT_EQ(10, pool_stats->agg_num_running_);
+
+ // Remove HOST_2 in Queue C.
+ delete_update = MakeTopicDelta(true);
+ delete_item.key = "POOL:" + QUEUE_C + "!" + HOST_2;
+ delete_item.deleted = true;
+ delete_update.topic_entries.push_back(delete_item);
+ StatestoreSubscriber::TopicDeltaMap delete_topic_deltas2;
+ delete_topic_deltas2.emplace(Statestore::IMPALA_REQUEST_QUEUE_TOPIC, delete_update);
+ admission_controller->UpdatePoolStats(delete_topic_deltas2, &outgoing_topic_updates);
+
+ ASSERT_EQ(3, admission_controller->host_stats_.size());
+ ASSERT_EQ(3000, admission_controller->host_stats_[HOST_2].mem_reserved);
+ ASSERT_EQ(0, admission_controller->host_stats_[HOST_1].mem_reserved);
+
+ // Remove HOST_2 in Queue B.
+ TTopicDelta delete_update3 = MakeTopicDelta(true);
+ TTopicItem delete_item3;
+ delete_item3.key = "POOL:" + QUEUE_B + "!" + HOST_2;
+ delete_item3.deleted = true;
+ delete_update3.topic_entries.push_back(delete_item3);
+ StatestoreSubscriber::TopicDeltaMap delete_topic_deltas3;
+ delete_topic_deltas3.emplace(Statestore::IMPALA_REQUEST_QUEUE_TOPIC, delete_update3);
+ admission_controller->UpdatePoolStats(delete_topic_deltas3, &outgoing_topic_updates);
+
+ ASSERT_EQ(3, admission_controller->host_stats_.size());
+ ASSERT_EQ(0, admission_controller->host_stats_[HOST_2].mem_reserved);
+ ASSERT_EQ(0, admission_controller->host_stats_[HOST_1].mem_reserved);
+}
+
/// Test CanAdmitRequest in the context of aggregated memory required to admit a query.
TEST_F(AdmissionControllerTest, CanAdmitRequestMemory) {
// Pass the paths of the configuration files as command line flags.
diff --git a/be/src/scheduling/admission-controller.cc b/be/src/scheduling/admission-controller.cc
index bb603383c..b274a2f2f 100644
--- a/be/src/scheduling/admission-controller.cc
+++ b/be/src/scheduling/admission-controller.cc
@@ -1714,6 +1714,7 @@ void AdmissionController::UpdatePoolStats(
StatestoreSubscriber::TopicDeltaMap::const_iterator topic =
incoming_topic_deltas.find(request_queue_topic_name_);
+ set<string> pool_stats_removed_hosts;
if (topic != incoming_topic_deltas.end()) {
const TTopicDelta& delta = topic->second;
// Delta and non-delta updates are handled the same way, except for a full update
@@ -1723,9 +1724,9 @@ void AdmissionController::UpdatePoolStats(
VLOG_ROW << "Full impala-request-queue stats update";
for (auto& entry : pool_stats_) entry.second.ClearRemoteStats();
}
- HandleTopicUpdates(delta.topic_entries);
+ HandleTopicUpdates(delta.topic_entries, pool_stats_removed_hosts);
}
- UpdateClusterAggregates();
+ UpdateClusterAggregates(pool_stats_removed_hosts);
last_topic_update_time_ms_ = MonotonicMillis();
pending_dequeue_ = true;
}
@@ -1755,7 +1756,8 @@ void AdmissionController::PoolStats::UpdateRemoteStats(const string& host_id,
}
}
-void AdmissionController::HandleTopicUpdates(const vector<TTopicItem>& topic_updates) {
+void AdmissionController::HandleTopicUpdates(
+ const vector<TTopicItem>& topic_updates, set<string>& pool_stats_removed_nodes) {
string topic_key_prefix;
string topic_key_suffix;
string pool_name;
@@ -1769,6 +1771,7 @@ void AdmissionController::HandleTopicUpdates(const vector<TTopicItem>& topic_upd
if (topic_backend_id == host_id_) continue;
if (item.deleted) {
GetPoolStats(pool_name)->UpdateRemoteStats(topic_backend_id, nullptr);
+ pool_stats_removed_nodes.insert(topic_backend_id);
continue;
}
TPoolStats remote_update;
@@ -1860,7 +1863,7 @@ void AdmissionController::PoolStats::UpdateAggregates(HostMemMap* host_mem_reser
VLOG_ROW << "Updated: " << DebugString();
}
-void AdmissionController::UpdateClusterAggregates() {
+void AdmissionController::UpdateClusterAggregates(const set<string>& removed_nodes) {
// Recompute mem_reserved for all hosts.
PoolStats::HostMemMap updated_mem_reserved;
for (auto& entry : pool_stats_) entry.second.UpdateAggregates(&updated_mem_reserved);
@@ -1878,6 +1881,26 @@ void AdmissionController::UpdateClusterAggregates() {
++i;
}
}
+
+ // We know if any host stats were removed from the pool stats during statestore topic
+ // update by the set pool_stats_removed_nodes. If a host was removed and no longer
+ // exists in any remote pool stats, we reset the mem_reserved to 0 for this host in
+ // the host stats.
+ for (const auto& host : removed_nodes) {
+ auto it = host_stats_.find(host);
+ if (it != host_stats_.end()
+ && updated_mem_reserved.find(host) == updated_mem_reserved.end()) {
+ int64_t old_mem_reserved = it->second.mem_reserved;
+ it->second.mem_reserved = 0;
+ if (VLOG_ROW_IS_ON) {
+ ss << endl
+ << "Mem_reserved reset to 0 for removed host: " << host
+ << " (old value=" << PrintBytes(old_mem_reserved) << ")";
+ ++i;
+ }
+ }
+ }
+
if (i > 0) VLOG_ROW << ss.str();
}
diff --git a/be/src/scheduling/admission-controller.h b/be/src/scheduling/admission-controller.h
index 296559c9c..ebc202faa 100644
--- a/be/src/scheduling/admission-controller.h
+++ b/be/src/scheduling/admission-controller.h
@@ -720,6 +720,7 @@ class AdmissionController {
FRIEND_TEST(AdmissionControllerTest, GetMaxToDequeue);
FRIEND_TEST(AdmissionControllerTest, QueryRejection);
FRIEND_TEST(AdmissionControllerTest, TopNQueryCheck);
+ FRIEND_TEST(AdmissionControllerTest, EraseHostStats);
friend class AdmissionControllerTest;
};
@@ -935,13 +936,18 @@ class AdmissionController {
/// Updates the remote stats with per-host topic_updates coming from the statestore.
/// Removes remote stats identified by topic deletions coming from the
/// statestore. Called by UpdatePoolStats(). Must hold admission_ctrl_lock_.
- void HandleTopicUpdates(const std::vector<TTopicItem>& topic_updates);
+ /// Any remote host stats removed from the pool stats during the process will result in
+ /// the host being added to the pool_stats_removed_nodes set.
+ void HandleTopicUpdates(const std::vector<TTopicItem>& topic_updates,
+ std::set<std::string>& pool_stats_removed_nodes);
/// Re-computes the per-pool aggregate stats and the per-host aggregates in host_stats_
/// using each pool's remote_stats_ and local_stats_.
/// Called by UpdatePoolStats() after handling updates and deletions.
/// Must hold admission_ctrl_lock_.
- void UpdateClusterAggregates();
+ /// If the host is present in removed_nodes and the host is not present in any remote
+ /// pool stats, the mem_reserved will be reset in the host_stats_ for that host.
+ void UpdateClusterAggregates(const std::set<std::string>& removed_nodes);
/// Computes schedules for all executor groups that can run the query in 'queue_node'.
/// For subsequent calls schedules are only re-computed if the membership version inside
@@ -1224,6 +1230,7 @@ class AdmissionController {
FRIEND_TEST(AdmissionControllerTest, DedicatedCoordScheduleState);
FRIEND_TEST(AdmissionControllerTest, DedicatedCoordAdmissionChecks);
FRIEND_TEST(AdmissionControllerTest, TopNQueryCheck);
+ FRIEND_TEST(AdmissionControllerTest, EraseHostStats);
friend class AdmissionControllerTest;
};