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/08/08 23:14:34 UTC
[impala] branch master updated (29a26a536 -> 308fda110)
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 29a26a536 IMPALA-11464: Skip listing staging dirs to avoid failures on them
new ba60c5f29 IMPALA-11401,IMPALA-10794: Add logs and thread names for catalogd RPCs
new cdf625968 IMPALA-11474: Codegen Tuple size in sorter-ir.cc
new 308fda110 IMPALA-11481: Make test_runtime_profile_aggregate slower to avoid flakyness
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/exec/partial-sort-node.cc | 4 +-
be/src/exec/sort-node.cc | 4 +-
be/src/exec/topn-node.cc | 3 +-
be/src/runtime/sorter-internal.h | 5 +-
be/src/runtime/sorter-ir.cc | 6 +-
be/src/runtime/sorter.cc | 6 +-
.../org/apache/impala/analysis/ColumnName.java | 6 +
.../org/apache/impala/analysis/FunctionName.java | 4 +
.../apache/impala/analysis/ResetMetadataStmt.java | 5 +
.../java/org/apache/impala/analysis/TableName.java | 4 +
.../impala/catalog/local/CatalogdMetaProvider.java | 4 +-
.../apache/impala/service/CatalogOpExecutor.java | 11 +-
.../java/org/apache/impala/service/JniCatalog.java | 146 +++++++++++---
.../java/org/apache/impala/util/CatalogOpUtil.java | 148 ++++++++++++++
.../org/apache/impala/util/CatalogOpUtilTest.java | 221 +++++++++++++++++++++
.../tpch/queries/runtime-profile-aggregated.test | 6 +-
16 files changed, 537 insertions(+), 46 deletions(-)
create mode 100644 fe/src/main/java/org/apache/impala/util/CatalogOpUtil.java
create mode 100644 fe/src/test/java/org/apache/impala/util/CatalogOpUtilTest.java
[impala] 03/03: IMPALA-11481: Make test_runtime_profile_aggregate slower to avoid flakyness
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 308fda110758b0fc58e5b1f477d635aac29aea75
Author: Csaba Ringhofer <cs...@cloudera.com>
AuthorDate: Mon Aug 8 17:15:11 2022 +0200
IMPALA-11481: Make test_runtime_profile_aggregate slower to avoid flakyness
The recent flakyness was probably caused by the new faster instances used
for testing. For more information see the Jira.
Change-Id: Ia31abfdcd12a26a2340c89d312b0d3b9c9fa3014
Reviewed-on: http://gerrit.cloudera.org:8080/18825
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
testdata/workloads/tpch/queries/runtime-profile-aggregated.test | 6 ++++--
1 file changed, 4 insertions(+), 2 deletions(-)
diff --git a/testdata/workloads/tpch/queries/runtime-profile-aggregated.test b/testdata/workloads/tpch/queries/runtime-profile-aggregated.test
index 14e4fb6d0..1966919e3 100644
--- a/testdata/workloads/tpch/queries/runtime-profile-aggregated.test
+++ b/testdata/workloads/tpch/queries/runtime-profile-aggregated.test
@@ -2,10 +2,12 @@
---- QUERY
# Sanity check aggregated runtime profile. Pick a query that runs for long enough for
# time series counters, etc to collect some data.
-select STRAIGHT_JOIN count(distinct l_partkey)
+select STRAIGHT_JOIN count(distinct l_partkey), count(distinct l_comment)
from lineitem join [BROADCAST] supplier on l_suppkey = s_suppkey
+---- TYPES
+BIGINT,BIGINT
---- RESULTS
-200000
+200000,4580667
---- RUNTIME_PROFILE
# Top-level counter (not aggregated)
row_regex: .*NumBackends: 4 .*
[impala] 01/03: IMPALA-11401,IMPALA-10794: Add logs and thread names for catalogd RPCs
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 ba60c5f29a3ed4164a9a08cdc816e291c4f85352
Author: stiga-huang <hu...@gmail.com>
AuthorDate: Fri Jul 22 15:34:05 2022 +0800
IMPALA-11401,IMPALA-10794: Add logs and thread names for catalogd RPCs
We've seen catalogd throws OutOfMemoryError when serializing large
responses (i.e. size > 2GB). However, the related table names are
missing in the logs. Admins would like to get the table names and
blacklist those tables until they are optimized (e.g. reducing
partitions).
To improve the supportability, this patch adds logs in the Catalogd RPC
code paths to log some details of the request, also add thread
annotations to improve readability of jstacks.
Tests:
- Add unit tests for short descriptions of requests.
- Manually add codes to throw OutOfMemoryError and verify the logs
shown as expected.
- Run test_concurrent_ddls.py and metadata tests. Capture jstacks and
verify the thread annotations are shown.
- Run CORE tests
Change-Id: Iac7f2eda8b95643a3d3c3bef64ea71b67b20595a
Reviewed-on: http://gerrit.cloudera.org:8080/18772
Reviewed-by: Csaba Ringhofer <cs...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
.../org/apache/impala/analysis/ColumnName.java | 6 +
.../org/apache/impala/analysis/FunctionName.java | 4 +
.../apache/impala/analysis/ResetMetadataStmt.java | 5 +
.../java/org/apache/impala/analysis/TableName.java | 4 +
.../impala/catalog/local/CatalogdMetaProvider.java | 4 +-
.../apache/impala/service/CatalogOpExecutor.java | 11 +-
.../java/org/apache/impala/service/JniCatalog.java | 146 +++++++++++---
.../java/org/apache/impala/util/CatalogOpUtil.java | 148 ++++++++++++++
.../org/apache/impala/util/CatalogOpUtilTest.java | 221 +++++++++++++++++++++
9 files changed, 513 insertions(+), 36 deletions(-)
diff --git a/fe/src/main/java/org/apache/impala/analysis/ColumnName.java b/fe/src/main/java/org/apache/impala/analysis/ColumnName.java
index 8eb285bce..41d50a1ab 100644
--- a/fe/src/main/java/org/apache/impala/analysis/ColumnName.java
+++ b/fe/src/main/java/org/apache/impala/analysis/ColumnName.java
@@ -18,6 +18,7 @@
package org.apache.impala.analysis;
import com.google.common.base.Preconditions;
+import org.apache.impala.thrift.TColumnName;
/**
* Represents a column name that optionally includes its database name.
@@ -36,4 +37,9 @@ public class ColumnName {
public TableName getTableName() { return tableName_; }
public String getColumnName() { return columnName_; }
+
+ public static String thriftToString(TColumnName colName) {
+ return TableName.thriftToString(colName.getTable_name()) + "." +
+ colName.getColumn_name();
+ }
}
diff --git a/fe/src/main/java/org/apache/impala/analysis/FunctionName.java b/fe/src/main/java/org/apache/impala/analysis/FunctionName.java
index 288f7cb6e..b3093c0c1 100644
--- a/fe/src/main/java/org/apache/impala/analysis/FunctionName.java
+++ b/fe/src/main/java/org/apache/impala/analysis/FunctionName.java
@@ -161,4 +161,8 @@ public class FunctionName {
public static FunctionName fromThrift(TFunctionName fnName) {
return new FunctionName(fnName.getDb_name(), fnName.getFunction_name());
}
+
+ public static String thriftToString(TFunctionName fnName) {
+ return fromThrift(fnName).toString();
+ }
}
diff --git a/fe/src/main/java/org/apache/impala/analysis/ResetMetadataStmt.java b/fe/src/main/java/org/apache/impala/analysis/ResetMetadataStmt.java
index 92d9f005d..73c78b335 100644
--- a/fe/src/main/java/org/apache/impala/analysis/ResetMetadataStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/ResetMetadataStmt.java
@@ -125,6 +125,11 @@ public class ResetMetadataStmt extends StatementBase {
@VisibleForTesting
protected Action getAction() { return action_; }
+ @VisibleForTesting
+ public void setRequestingUser(User user) {
+ requestingUser_ = user;
+ }
+
@Override
public void collectTableRefs(List<TableRef> tblRefs) {
if (tableName_ != null && partitionSpec_ != null) {
diff --git a/fe/src/main/java/org/apache/impala/analysis/TableName.java b/fe/src/main/java/org/apache/impala/analysis/TableName.java
index ae6426e33..e413cca2b 100644
--- a/fe/src/main/java/org/apache/impala/analysis/TableName.java
+++ b/fe/src/main/java/org/apache/impala/analysis/TableName.java
@@ -126,6 +126,10 @@ public class TableName {
return new TableName(tableName.getDb_name(), tableName.getTable_name());
}
+ public static String thriftToString(TTableName tableName) {
+ return fromThrift(tableName).toString();
+ }
+
public TTableName toThrift() { return new TTableName(db_, tbl_); }
/**
diff --git a/fe/src/main/java/org/apache/impala/catalog/local/CatalogdMetaProvider.java b/fe/src/main/java/org/apache/impala/catalog/local/CatalogdMetaProvider.java
index 330161c4c..f43686b81 100644
--- a/fe/src/main/java/org/apache/impala/catalog/local/CatalogdMetaProvider.java
+++ b/fe/src/main/java/org/apache/impala/catalog/local/CatalogdMetaProvider.java
@@ -97,6 +97,7 @@ import org.apache.impala.thrift.TValidWriteIdList;
import org.apache.impala.util.AcidUtils;
import org.apache.impala.util.IcebergUtil;
import org.apache.impala.util.ListMap;
+import org.apache.impala.util.ThreadNameAnnotator;
import org.apache.thrift.TDeserializer;
import org.apache.thrift.TException;
import org.apache.thrift.TSerializer;
@@ -509,7 +510,8 @@ public class CatalogdMetaProvider implements MetaProvider {
Stopwatch sw = Stopwatch.createStarted();
boolean hit = false;
boolean isPiggybacked = false;
- try {
+ try (ThreadNameAnnotator tna = new ThreadNameAnnotator(
+ "LoadWithCaching for " + itemString)) {
CompletableFuture<Object> f = new CompletableFuture<Object>();
// NOTE: the Cache ensures that this is an atomic operation of either returning
// an existing value or inserting our own. Only one thread can think it is the
diff --git a/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java b/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
index 256301654..d641b90ca 100644
--- a/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
+++ b/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
@@ -232,6 +232,7 @@ import org.apache.impala.thrift.TUpdateCatalogResponse;
import org.apache.impala.thrift.TUpdatedPartition;
import org.apache.impala.util.AcidUtils;
import org.apache.impala.util.AcidUtils.TblTransaction;
+import org.apache.impala.util.CatalogOpUtil;
import org.apache.impala.util.CompressionUtil;
import org.apache.impala.util.DebugUtils;
import org.apache.impala.util.HdfsCachingUtil;
@@ -239,6 +240,7 @@ import org.apache.impala.util.IcebergUtil;
import org.apache.impala.util.KuduUtil;
import org.apache.impala.util.MetaStoreUtil;
import org.apache.impala.util.MetaStoreUtil.TableInsertEventInfo;
+import org.apache.impala.util.ThreadNameAnnotator;
import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -5506,7 +5508,10 @@ public class CatalogOpExecutor {
// Add partitions to metastore.
Map<String, Long> partitionToEventId = Maps.newHashMap();
- try (MetaStoreClient msClient = catalog_.getMetaStoreClient()) {
+ String annotation = String.format("Recovering %d partitions for %s",
+ hmsPartitions.size(), tbl.getFullName());
+ try (ThreadNameAnnotator tna = new ThreadNameAnnotator(annotation);
+ MetaStoreClient msClient = catalog_.getMetaStoreClient()) {
List<Partition> addedPartitions = addHmsPartitions(msClient, tbl, hmsPartitions,
partitionToEventId, true);
addHdfsPartitions(msClient, tbl, addedPartitions, partitionToEventId);
@@ -6045,9 +6050,7 @@ public class CatalogOpExecutor {
*/
public TResetMetadataResponse execResetMetadata(TResetMetadataRequest req)
throws CatalogException {
- String cmdString = String.format("%s issued by %s",
- req.is_refresh ? "REFRESH":"INVALIDATE",
- req.header != null ? req.header.requesting_user : " unknown user");
+ String cmdString = CatalogOpUtil.getShortDescForReset(req);
TResetMetadataResponse resp = new TResetMetadataResponse();
resp.setResult(new TCatalogUpdateResult());
resp.getResult().setCatalog_service_id(JniCatalog.getServiceId());
diff --git a/fe/src/main/java/org/apache/impala/service/JniCatalog.java b/fe/src/main/java/org/apache/impala/service/JniCatalog.java
index e8f2951f4..126d08f63 100644
--- a/fe/src/main/java/org/apache/impala/service/JniCatalog.java
+++ b/fe/src/main/java/org/apache/impala/service/JniCatalog.java
@@ -30,6 +30,7 @@ import org.apache.hadoop.hive.metastore.api.CurrentNotificationEventId;
import org.apache.impala.authorization.AuthorizationConfig;
import org.apache.impala.authorization.AuthorizationFactory;
import org.apache.impala.authorization.AuthorizationManager;
+import org.apache.impala.catalog.Catalog;
import org.apache.impala.catalog.CatalogException;
import org.apache.impala.catalog.CatalogServiceCatalog;
import org.apache.impala.catalog.Db;
@@ -46,6 +47,7 @@ import org.apache.impala.catalog.metastore.ICatalogMetastoreServer;
import org.apache.impala.catalog.metastore.NoOpCatalogMetastoreServer;
import org.apache.impala.catalog.monitor.CatalogMonitor;
import org.apache.impala.catalog.monitor.CatalogOperationMetrics;
+import org.apache.impala.common.PrintUtils;
import org.apache.impala.compat.MetastoreShim;
import org.apache.impala.common.ImpalaException;
import org.apache.impala.common.InternalException;
@@ -78,8 +80,10 @@ import org.apache.impala.thrift.TUpdateCatalogRequest;
import org.apache.impala.thrift.TBackendGflags;
import org.apache.impala.thrift.TUpdateTableUsageRequest;
import org.apache.impala.util.AuthorizationUtil;
+import org.apache.impala.util.CatalogOpUtil;
import org.apache.impala.util.GlogAppender;
import org.apache.impala.util.PatternMatcher;
+import org.apache.impala.util.ThreadNameAnnotator;
import org.apache.thrift.TException;
import org.apache.thrift.TSerializer;
import org.apache.thrift.protocol.TBinaryProtocol;
@@ -219,11 +223,20 @@ public class JniCatalog {
long start = System.currentTimeMillis();
TGetCatalogDeltaRequest params = new TGetCatalogDeltaRequest();
JniUtil.deserializeThrift(protocolFactory_, params, thriftGetCatalogDeltaReq);
- byte[] res = new TSerializer(protocolFactory_).serialize(new TGetCatalogDeltaResponse(
- catalog_.getCatalogDelta(params.getNative_catalog_server_ptr(),
- params.getFrom_version())));
- JniUtil.logResponse(res.length, start, params, "getCatalogDelta");
- return res;
+ TSerializer serializer = new TSerializer(protocolFactory_);
+ String shortDesc = "getting catalog delta from version " + params.getFrom_version();
+ try (ThreadNameAnnotator tna = new ThreadNameAnnotator(shortDesc)) {
+ byte[] res = serializer.serialize(new TGetCatalogDeltaResponse(
+ catalog_.getCatalogDelta(params.getNative_catalog_server_ptr(),
+ params.getFrom_version())));
+ JniUtil.logResponse(res.length, start, params, "getCatalogDelta");
+ return res;
+ } catch (Throwable e) {
+ long duration = System.currentTimeMillis() - start;
+ LOG.error("Error in {}. Time spent: {}.",
+ shortDesc, PrintUtils.printTimeMs(duration));
+ throw e;
+ }
}
/**
@@ -236,17 +249,25 @@ public class JniCatalog {
/**
* Executes the given DDL request and returns the result.
*/
- public byte[] execDdl(byte[] thriftDdlExecReq) throws ImpalaException {
+ public byte[] execDdl(byte[] thriftDdlExecReq) throws ImpalaException, TException {
long start = System.currentTimeMillis();
TDdlExecRequest params = new TDdlExecRequest();
JniUtil.deserializeThrift(protocolFactory_, params, thriftDdlExecReq);
TSerializer serializer = new TSerializer(protocolFactory_);
- try {
+ String shortDesc = CatalogOpUtil.getShortDescForExecDdl(params);
+ LOG.info("execDdl request: " + shortDesc);
+ try (ThreadNameAnnotator tna = new ThreadNameAnnotator(shortDesc)) {
byte[] res = serializer.serialize(catalogOpExecutor_.execDdlRequest(params));
JniUtil.logResponse(res.length, start, params, "execDdl");
+ long duration = System.currentTimeMillis() - start;
+ LOG.info("finished execDdl request: {}. Time spent: {}",
+ shortDesc, PrintUtils.printTimeMs(duration));
return res;
- } catch (TException e) {
- throw new InternalException(e.getMessage());
+ } catch (Throwable e) {
+ long duration = System.currentTimeMillis() - start;
+ LOG.error("Error in execDdl for {}. Time spent: {}.",
+ shortDesc, PrintUtils.printTimeMs(duration));
+ throw e;
}
}
@@ -260,10 +281,20 @@ public class JniCatalog {
JniUtil.deserializeThrift(protocolFactory_, req, thriftResetMetadataReq);
TSerializer serializer = new TSerializer(protocolFactory_);
catalogOperationUsage.increment(req);
- try {
+ String shortDesc = CatalogOpUtil.getShortDescForReset(req);
+ LOG.info("resetMetadata request: " + shortDesc);
+ try (ThreadNameAnnotator tna = new ThreadNameAnnotator(shortDesc)) {
byte[] res = serializer.serialize(catalogOpExecutor_.execResetMetadata(req));
JniUtil.logResponse(res.length, start, req, "resetMetadata");
+ long duration = System.currentTimeMillis() - start;
+ LOG.info("finished resetMetadata request: {}. Time spent: {}",
+ shortDesc, PrintUtils.printTimeMs(duration));
return res;
+ } catch (Throwable e) {
+ long duration = System.currentTimeMillis() - start;
+ LOG.error("Error in resetMetadata for {}. Time spent: {}.",
+ shortDesc, PrintUtils.printTimeMs(duration));
+ throw e;
} finally {
catalogOperationUsage.decrement(req);
}
@@ -317,9 +348,17 @@ public class JniCatalog {
long start = System.currentTimeMillis();
TGetTableMetricsParams params = new TGetTableMetricsParams();
JniUtil.deserializeThrift(protocolFactory_, params, getTableMetricsParams);
- String res = catalog_.getTableMetrics(params.table_name);
- JniUtil.logResponse(res.length(), start, params, "getTableMetrics");
- return res;
+ try (ThreadNameAnnotator tna = new ThreadNameAnnotator(
+ "getTableMetrics " + params.table_name)) {
+ String res = catalog_.getTableMetrics(params.table_name);
+ JniUtil.logResponse(res.length(), start, params, "getTableMetrics");
+ return res;
+ } catch (Throwable e) {
+ long duration = System.currentTimeMillis() - start;
+ LOG.error("Error in getTableMetrics {}. Time spent: {}.", params.table_name,
+ PrintUtils.printTimeMs(duration));
+ throw e;
+ }
}
/**
@@ -328,12 +367,21 @@ public class JniCatalog {
public byte[] getCatalogObject(byte[] thriftParams) throws ImpalaException,
TException {
long start = System.currentTimeMillis();
- TCatalogObject objectDescription = new TCatalogObject();
- JniUtil.deserializeThrift(protocolFactory_, objectDescription, thriftParams);
+ TCatalogObject objectDesc = new TCatalogObject();
+ JniUtil.deserializeThrift(protocolFactory_, objectDesc, thriftParams);
TSerializer serializer = new TSerializer(protocolFactory_);
- byte[] res = serializer.serialize(catalog_.getTCatalogObject(objectDescription));
- JniUtil.logResponse(res.length, start, objectDescription, "getCatalogObject");
- return res;
+ String shortDesc = "getting thrift catalog object of "
+ + Catalog.toCatalogObjectKey(objectDesc);
+ try (ThreadNameAnnotator tna = new ThreadNameAnnotator(shortDesc)) {
+ byte[] res = serializer.serialize(catalog_.getTCatalogObject(objectDesc));
+ JniUtil.logResponse(res.length, start, objectDesc, "getCatalogObject");
+ return res;
+ } catch (Throwable e) {
+ long duration = System.currentTimeMillis() - start;
+ LOG.error("Error in {}. Time spent: {}.", shortDesc,
+ PrintUtils.printTimeMs(duration));
+ throw e;
+ }
}
/**
@@ -343,12 +391,21 @@ public class JniCatalog {
public String getJsonCatalogObject(byte[] thriftParams) throws ImpalaException,
TException {
long start = System.currentTimeMillis();
- TCatalogObject objectDescription = new TCatalogObject();
- JniUtil.deserializeThrift(protocolFactory_, objectDescription, thriftParams);
+ TCatalogObject objectDesc = new TCatalogObject();
+ JniUtil.deserializeThrift(protocolFactory_, objectDesc, thriftParams);
TSerializer jsonSerializer = new TSerializer(new TSimpleJSONProtocol.Factory());
- String res = jsonSerializer.toString(catalog_.getTCatalogObject(objectDescription));
- JniUtil.logResponse(res.length(), start, objectDescription, "getJsonCatalogObject");
- return res;
+ String shortDesc = "getting json catalog object of "
+ + Catalog.toCatalogObjectKey(objectDesc);
+ try (ThreadNameAnnotator tna = new ThreadNameAnnotator(shortDesc)) {
+ String res = jsonSerializer.toString(catalog_.getTCatalogObject(objectDesc));
+ JniUtil.logResponse(res.length(), start, objectDesc, "getJsonCatalogObject");
+ return res;
+ } catch (Throwable e) {
+ long duration = System.currentTimeMillis() - start;
+ LOG.error("Error in {}. Time spent: {}.", shortDesc,
+ PrintUtils.printTimeMs(duration));
+ throw e;
+ }
}
public byte[] getPartialCatalogObject(byte[] thriftParams) throws ImpalaException,
@@ -358,9 +415,16 @@ public class JniCatalog {
new TGetPartialCatalogObjectRequest();
JniUtil.deserializeThrift(protocolFactory_, req, thriftParams);
TSerializer serializer = new TSerializer(protocolFactory_);
- byte[] res = serializer.serialize(catalog_.getPartialCatalogObject(req));
- JniUtil.logResponse(res.length, start, req, "getPartialCatalogObject");
- return res;
+ try {
+ byte[] res = serializer.serialize(catalog_.getPartialCatalogObject(req));
+ JniUtil.logResponse(res.length, start, req, "getPartialCatalogObject");
+ return res;
+ } catch (Throwable e) {
+ long duration = System.currentTimeMillis() - start;
+ LOG.error("Error in getting PartialCatalogObject of {}. Time spent: {}.",
+ Catalog.toCatalogObjectKey(req.object_desc), PrintUtils.printTimeMs(duration));
+ throw e;
+ }
}
/**
@@ -405,15 +469,24 @@ public class JniCatalog {
JniUtil.deserializeThrift(protocolFactory_, request, thriftParams);
TSerializer serializer = new TSerializer(protocolFactory_);
TGetPartitionStatsResponse response = new TGetPartitionStatsResponse();
- try {
+ try (ThreadNameAnnotator tna = new ThreadNameAnnotator(
+ "Getting partition stats of " + request.table_name)) {
response.setPartition_stats(catalog_.getPartitionStats(request));
} catch (CatalogException e) {
response.setStatus(
new TStatus(TErrorCode.INTERNAL_ERROR, ImmutableList.of(e.getMessage())));
}
- byte[] res = serializer.serialize(response);
- JniUtil.logResponse(res.length, start, request, "getPartitionStats");
- return res;
+ try (ThreadNameAnnotator tna = new ThreadNameAnnotator(
+ "Serializing partition stats of " + request.table_name)) {
+ byte[] res = serializer.serialize(response);
+ JniUtil.logResponse(res.length, start, request, "getPartitionStats");
+ return res;
+ } catch (Throwable e) {
+ long duration = System.currentTimeMillis() - start;
+ LOG.error("Error in serializing partition stats of {}. Time spent in method: {}.",
+ request.table_name, PrintUtils.printTimeMs(duration));
+ throw e;
+ }
}
/**
@@ -427,10 +500,21 @@ public class JniCatalog {
JniUtil.deserializeThrift(protocolFactory_, request, thriftUpdateCatalog);
TSerializer serializer = new TSerializer(protocolFactory_);
catalogOperationUsage.increment(request);
- try {
+ String shortDesc = String.format("updateCatalog for %s.%s",
+ request.db_name, request.target_table);
+ LOG.info(shortDesc);
+ try (ThreadNameAnnotator tna = new ThreadNameAnnotator(shortDesc)) {
byte[] res = serializer.serialize(catalogOpExecutor_.updateCatalog(request));
JniUtil.logResponse(res.length, start, request, "updateCatalog");
+ long duration = System.currentTimeMillis() - start;
+ LOG.info("finished {}. Time spent: {}", shortDesc,
+ PrintUtils.printTimeMs(duration));
return res;
+ } catch (Throwable e) {
+ long duration = System.currentTimeMillis() - start;
+ LOG.error("Error in {}. Time spent: {}.", shortDesc,
+ PrintUtils.printTimeMs(duration));
+ throw e;
} finally {
catalogOperationUsage.decrement(request);
}
diff --git a/fe/src/main/java/org/apache/impala/util/CatalogOpUtil.java b/fe/src/main/java/org/apache/impala/util/CatalogOpUtil.java
new file mode 100644
index 000000000..0b96a3ecb
--- /dev/null
+++ b/fe/src/main/java/org/apache/impala/util/CatalogOpUtil.java
@@ -0,0 +1,148 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.impala.util;
+
+import org.apache.impala.analysis.ColumnName;
+import org.apache.impala.analysis.FunctionName;
+import org.apache.impala.analysis.TableName;
+import org.apache.impala.thrift.TCommentOnParams;
+import org.apache.impala.thrift.TDdlExecRequest;
+import org.apache.impala.thrift.TResetMetadataRequest;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class CatalogOpUtil {
+ private static final Logger LOG = LoggerFactory.getLogger(CatalogOpUtil.class);
+
+ /**
+ * Get a short description for the execDdl request.
+ */
+ public static String getShortDescForExecDdl(TDdlExecRequest req) {
+ String target;
+ try {
+ switch (req.ddl_type) {
+ case ALTER_DATABASE:
+ target = req.getAlter_db_params().getDb();
+ break;
+ case ALTER_TABLE:
+ target = TableName.thriftToString(req.getAlter_table_params().getTable_name());
+ break;
+ case ALTER_VIEW:
+ target = TableName.thriftToString(req.getAlter_view_params().getView_name());
+ break;
+ case CREATE_DATABASE:
+ target = req.getCreate_db_params().getDb();
+ break;
+ case CREATE_TABLE_AS_SELECT:
+ case CREATE_TABLE:
+ target = TableName.thriftToString(req.getCreate_table_params().getTable_name());
+ break;
+ case CREATE_TABLE_LIKE:
+ target = TableName.thriftToString(
+ req.getCreate_table_like_params().getTable_name());
+ break;
+ case CREATE_VIEW:
+ target = TableName.thriftToString(req.getCreate_view_params().getView_name());
+ break;
+ case CREATE_FUNCTION:
+ target = FunctionName.thriftToString(
+ req.getCreate_fn_params().getFn().getName());
+ break;
+ case COMMENT_ON: {
+ TCommentOnParams params = req.getComment_on_params();
+ if (params.isSetDb()) {
+ target = "DB " + params.getDb();
+ } else if (params.isSetTable_name()) {
+ target = "TABLE " + TableName.thriftToString(params.getTable_name());
+ } else if (params.isSetColumn_name()) {
+ target = "COLUMN " + ColumnName.thriftToString(params.getColumn_name());
+ } else {
+ target = "";
+ }
+ break;
+ }
+ case DROP_STATS:
+ target = TableName.thriftToString(req.getDrop_stats_params().getTable_name());
+ break;
+ case DROP_DATABASE:
+ target = req.getDrop_db_params().getDb();
+ break;
+ case DROP_TABLE:
+ case DROP_VIEW:
+ target = TableName.thriftToString(
+ req.getDrop_table_or_view_params().getTable_name());
+ break;
+ case TRUNCATE_TABLE:
+ target = TableName.thriftToString(req.getTruncate_params().getTable_name());
+ break;
+ case DROP_FUNCTION:
+ target = FunctionName.thriftToString(req.getDrop_fn_params().fn_name);
+ break;
+ case CREATE_ROLE:
+ case DROP_ROLE:
+ target = req.getCreate_drop_role_params().getRole_name();
+ break;
+ case GRANT_ROLE:
+ case REVOKE_ROLE:
+ target = req.getGrant_revoke_role_params().getRole_names() + " GROUP " +
+ req.getGrant_revoke_role_params().getGroup_names();
+ break;
+ case GRANT_PRIVILEGE:
+ target = "TO " + req.getGrant_revoke_priv_params().getPrincipal_name();
+ break;
+ case REVOKE_PRIVILEGE:
+ target = "FROM " + req.getGrant_revoke_priv_params().getPrincipal_name();
+ break;
+ default:
+ target = "";
+ }
+ } catch (Throwable t) {
+ // This method is used for all DDL RPCs. We should not fail them by errors happen
+ // here. Catch all exceptions and just log the error.
+ target = "unknown target";
+ LOG.error("Failed to get the target for request", t);
+ }
+
+ String user = "unknown user";
+ if (req.isSetHeader() && req.header.isSetRequesting_user()) {
+ user = req.header.requesting_user;
+ }
+ return String.format("%s%s issued by %s", req.ddl_type, " " + target, user);
+ }
+
+ /**
+ * Get a short description for the resetMetadata request.
+ */
+ public static String getShortDescForReset(TResetMetadataRequest req) {
+ String cmd = req.is_refresh ? "REFRESH " : "INVALIDATE ";
+ if (req.isSetDb_name()) {
+ if (req.is_refresh) cmd += "FUNCTIONS IN ";
+ cmd += "DATABASE " + req.getDb_name();
+ } else if (req.isSetTable_name()) {
+ cmd += "TABLE " + TableName.fromThrift(req.getTable_name());
+ if (req.isSetPartition_spec()) cmd += " PARTITIONS";
+ } else if (req.isAuthorization()) {
+ cmd += "AUTHORIZATION";
+ } else {
+ // Global INVALIDATE METADATA
+ cmd += "ALL";
+ }
+ String user = req.header != null ? req.header.requesting_user : "unknown user";
+ return String.format("%s issued by %s", cmd, user);
+ }
+}
diff --git a/fe/src/test/java/org/apache/impala/util/CatalogOpUtilTest.java b/fe/src/test/java/org/apache/impala/util/CatalogOpUtilTest.java
new file mode 100644
index 000000000..11d8a9f62
--- /dev/null
+++ b/fe/src/test/java/org/apache/impala/util/CatalogOpUtilTest.java
@@ -0,0 +1,221 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.impala.util;
+
+import org.apache.impala.analysis.GrantRevokePrivStmt;
+import org.apache.impala.analysis.GrantRevokeRoleStmt;
+import org.apache.impala.analysis.PrivilegeSpec;
+import org.apache.impala.analysis.ResetMetadataStmt;
+import org.apache.impala.analysis.TableName;
+import org.apache.impala.authorization.User;
+import org.apache.impala.thrift.TAlterDbParams;
+import org.apache.impala.thrift.TAlterDbType;
+import org.apache.impala.thrift.TAlterTableParams;
+import org.apache.impala.thrift.TColumnName;
+import org.apache.impala.thrift.TCommentOnParams;
+import org.apache.impala.thrift.TCreateDbParams;
+import org.apache.impala.thrift.TCreateDropRoleParams;
+import org.apache.impala.thrift.TCreateFunctionParams;
+import org.apache.impala.thrift.TCreateOrAlterViewParams;
+import org.apache.impala.thrift.TCreateTableParams;
+import org.apache.impala.thrift.TDdlExecRequest;
+import org.apache.impala.thrift.TDdlQueryOptions;
+import org.apache.impala.thrift.TDdlType;
+import org.apache.impala.thrift.TDropDbParams;
+import org.apache.impala.thrift.TDropFunctionParams;
+import org.apache.impala.thrift.TDropTableOrViewParams;
+import org.apache.impala.thrift.TFunction;
+import org.apache.impala.thrift.TFunctionName;
+import org.apache.impala.thrift.TGrantRevokePrivParams;
+import org.apache.impala.thrift.TGrantRevokeRoleParams;
+import org.apache.impala.thrift.TPrincipalType;
+import org.apache.impala.thrift.TPrivilegeLevel;
+import org.apache.impala.thrift.TResetMetadataRequest;
+import org.apache.impala.thrift.TTableName;
+import org.junit.Test;
+
+import java.util.Collections;
+
+import static org.junit.Assert.assertEquals;
+
+public class CatalogOpUtilTest {
+
+ private void testResetStmt(ResetMetadataStmt stmt, User user, String expected)
+ throws Exception {
+ stmt.setRequestingUser(user);
+ assertEquals(expected, CatalogOpUtil.getShortDescForReset(stmt.toThrift()));
+ }
+
+ @Test
+ public void testResetMetadataDesc() throws Exception {
+ User user = new User("Alice");
+ TableName tblName = new TableName("default", "tbl");
+
+ testResetStmt(ResetMetadataStmt.createInvalidateStmt(),
+ user, "INVALIDATE ALL issued by Alice");
+ testResetStmt(ResetMetadataStmt.createInvalidateStmt(tblName),
+ user, "INVALIDATE TABLE default.tbl issued by Alice");
+ testResetStmt(ResetMetadataStmt.createRefreshTableStmt(tblName),
+ user, "REFRESH TABLE default.tbl issued by Alice");
+ testResetStmt(ResetMetadataStmt.createRefreshFunctionsStmt("db1"),
+ user, "REFRESH FUNCTIONS IN DATABASE db1 issued by Alice");
+ testResetStmt(ResetMetadataStmt.createRefreshAuthorizationStmt(),
+ user, "REFRESH AUTHORIZATION issued by Alice");
+
+ // Test REFRESH PARTITIONS using a fake partition spec
+ ResetMetadataStmt stmt = ResetMetadataStmt.createRefreshTableStmt(tblName);
+ stmt.setRequestingUser(user);
+ TResetMetadataRequest req = stmt.toThrift();
+ req.setPartition_spec(Collections.emptyList());
+ assertEquals("REFRESH TABLE default.tbl PARTITIONS issued by Alice",
+ CatalogOpUtil.getShortDescForReset(req));
+ }
+
+ @Test
+ public void testDdlDesc() {
+ TDdlExecRequest req;
+ TTableName tblName = new TTableName("db1", "tbl1");
+
+ req = new TDdlExecRequest();
+ req.setQuery_options(new TDdlQueryOptions());
+ req.setDdl_type(TDdlType.CREATE_DATABASE);
+ TCreateDbParams createDbParams = new TCreateDbParams();
+ createDbParams.setDb("db1");
+ req.setCreate_db_params(createDbParams);
+ assertEquals("CREATE_DATABASE db1 issued by unknown user",
+ CatalogOpUtil.getShortDescForExecDdl(req));
+
+ req.setDdl_type(TDdlType.DROP_DATABASE);
+ TDropDbParams dropDbParams = new TDropDbParams();
+ dropDbParams.setDb("db1");
+ req.setDrop_db_params(dropDbParams);
+ assertEquals("DROP_DATABASE db1 issued by unknown user",
+ CatalogOpUtil.getShortDescForExecDdl(req));
+
+ req.setDdl_type(TDdlType.ALTER_DATABASE);
+ TAlterDbParams alterDbParams = new TAlterDbParams();
+ alterDbParams.setAlter_type(TAlterDbType.SET_OWNER);
+ alterDbParams.setDb("db1");
+ req.setAlter_db_params(alterDbParams);
+ assertEquals("ALTER_DATABASE db1 issued by unknown user",
+ CatalogOpUtil.getShortDescForExecDdl(req));
+
+ req.setDdl_type(TDdlType.CREATE_TABLE);
+ TCreateTableParams createTableParams = new TCreateTableParams();
+ createTableParams.setTable_name(tblName);
+ req.setCreate_table_params(createTableParams);
+ assertEquals("CREATE_TABLE db1.tbl1 issued by unknown user",
+ CatalogOpUtil.getShortDescForExecDdl(req));
+ req.setDdl_type(TDdlType.CREATE_TABLE_AS_SELECT);
+ assertEquals("CREATE_TABLE_AS_SELECT db1.tbl1 issued by unknown user",
+ CatalogOpUtil.getShortDescForExecDdl(req));
+
+ req.setDdl_type(TDdlType.ALTER_TABLE);
+ TAlterTableParams alterTableParams = new TAlterTableParams();
+ alterTableParams.setTable_name(tblName);
+ req.setAlter_table_params(alterTableParams);
+ assertEquals("ALTER_TABLE db1.tbl1 issued by unknown user",
+ CatalogOpUtil.getShortDescForExecDdl(req));
+
+ req.setDdl_type(TDdlType.CREATE_VIEW);
+ TCreateOrAlterViewParams alterViewParams = new TCreateOrAlterViewParams();
+ alterViewParams.setView_name(tblName);
+ req.setCreate_view_params(alterViewParams);
+ assertEquals("CREATE_VIEW db1.tbl1 issued by unknown user",
+ CatalogOpUtil.getShortDescForExecDdl(req));
+ req.setDdl_type(TDdlType.ALTER_VIEW);
+ req.setAlter_view_params(alterViewParams);
+ assertEquals("ALTER_VIEW db1.tbl1 issued by unknown user",
+ CatalogOpUtil.getShortDescForExecDdl(req));
+
+ req.setDdl_type(TDdlType.DROP_TABLE);
+ TDropTableOrViewParams dropTableOrViewParams = new TDropTableOrViewParams();
+ dropTableOrViewParams.setTable_name(tblName);
+ req.setDrop_table_or_view_params(dropTableOrViewParams);
+ assertEquals("DROP_TABLE db1.tbl1 issued by unknown user",
+ CatalogOpUtil.getShortDescForExecDdl(req));
+ req.setDdl_type(TDdlType.DROP_VIEW);
+ assertEquals("DROP_VIEW db1.tbl1 issued by unknown user",
+ CatalogOpUtil.getShortDescForExecDdl(req));
+
+ req.setDdl_type(TDdlType.COMMENT_ON);
+ TCommentOnParams commentOnParams = new TCommentOnParams();
+ commentOnParams.setDb("db1");
+ req.setComment_on_params(commentOnParams);
+ assertEquals("COMMENT_ON DB db1 issued by unknown user",
+ CatalogOpUtil.getShortDescForExecDdl(req));
+ commentOnParams.clear();
+ commentOnParams.setTable_name(tblName);
+ req.setComment_on_params(commentOnParams);
+ assertEquals("COMMENT_ON TABLE db1.tbl1 issued by unknown user",
+ CatalogOpUtil.getShortDescForExecDdl(req));
+ commentOnParams.clear();
+ commentOnParams.setColumn_name(new TColumnName(tblName, "col1"));
+ req.setComment_on_params(commentOnParams);
+ assertEquals("COMMENT_ON COLUMN db1.tbl1.col1 issued by unknown user",
+ CatalogOpUtil.getShortDescForExecDdl(req));
+
+ req.setDdl_type(TDdlType.CREATE_FUNCTION);
+ TCreateFunctionParams createFunctionParams = new TCreateFunctionParams();
+ TFunction fn = new TFunction();
+ fn.setName(new TFunctionName("my_func"));
+ createFunctionParams.setFn(fn);
+ req.setCreate_fn_params(createFunctionParams);
+ assertEquals("CREATE_FUNCTION my_func issued by unknown user",
+ CatalogOpUtil.getShortDescForExecDdl(req));
+
+ req.setDdl_type(TDdlType.DROP_FUNCTION);
+ TDropFunctionParams dropFunctionParams = new TDropFunctionParams();
+ dropFunctionParams.setFn_name(new TFunctionName("my_func"));
+ req.setDrop_fn_params(dropFunctionParams);
+ assertEquals("DROP_FUNCTION my_func issued by unknown user",
+ CatalogOpUtil.getShortDescForExecDdl(req));
+
+ req.setDdl_type(TDdlType.CREATE_ROLE);
+ TCreateDropRoleParams createDropRoleParams = new TCreateDropRoleParams();
+ createDropRoleParams.setRole_name("my_role");
+ req.setCreate_drop_role_params(createDropRoleParams);
+ assertEquals("CREATE_ROLE my_role issued by unknown user",
+ CatalogOpUtil.getShortDescForExecDdl(req));
+ req.setDdl_type(TDdlType.DROP_ROLE);
+ assertEquals("DROP_ROLE my_role issued by unknown user",
+ CatalogOpUtil.getShortDescForExecDdl(req));
+
+ req.setDdl_type(TDdlType.GRANT_ROLE);
+ TGrantRevokeRoleParams grantRevokeRoleParams;
+ grantRevokeRoleParams = new GrantRevokeRoleStmt("my_role", "my_group", true)
+ .toThrift();
+ req.setGrant_revoke_role_params(grantRevokeRoleParams);
+ assertEquals("GRANT_ROLE [my_role] GROUP [my_group] issued by unknown user",
+ CatalogOpUtil.getShortDescForExecDdl(req));
+ req.setDdl_type(TDdlType.REVOKE_ROLE);
+ assertEquals("REVOKE_ROLE [my_role] GROUP [my_group] issued by unknown user",
+ CatalogOpUtil.getShortDescForExecDdl(req));
+
+ req.setDdl_type(TDdlType.GRANT_PRIVILEGE);
+ TGrantRevokePrivParams grantRevokePrivParams = new GrantRevokePrivStmt("my_role",
+ PrivilegeSpec.createServerScopedPriv(TPrivilegeLevel.SELECT), true, false,
+ TPrincipalType.ROLE).toThrift();
+ req.setGrant_revoke_priv_params(grantRevokePrivParams);
+ assertEquals("GRANT_PRIVILEGE TO my_role issued by unknown user",
+ CatalogOpUtil.getShortDescForExecDdl(req));
+ req.setDdl_type(TDdlType.REVOKE_PRIVILEGE);
+ assertEquals("REVOKE_PRIVILEGE FROM my_role issued by unknown user",
+ CatalogOpUtil.getShortDescForExecDdl(req));
+ }
+}
[impala] 02/03: IMPALA-11474: Codegen Tuple size in sorter-ir.cc
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 cdf62596846916e64ff9b0549b1eee8736185687
Author: noemi <np...@cloudera.com>
AuthorDate: Mon Jul 18 17:40:31 2022 +0200
IMPALA-11474: Codegen Tuple size in sorter-ir.cc
The number of bytes in a tuple is known before execution, it is
available in the query plan. However, currently the tuple size
is treated as a member variable of the TupleSorter.
Using Codegen to replace this member variable by a constant in
sorter-ir.cc can speed up the quicksort phase by up to 20% in the
most simple queries with small tuples.
Some examples using tpch_parquet lineitem, scale factor=8:
disable_outermost_topn=1;
Query: select _ from lineitem order by _ limit 1;
+---------------+-------------------------+------+----------+----------+-------------+
| Order by | Tuple | NDV | Constant | Variable | Improvement |
+---------------+-------------------------+------+----------+----------+-------------+
| rand() | int, rand | 48M | 12.21s | 14.21s | 14% |
| l_linenumber | int | 7 | 1.81s | 2.34s | 22% |
| l_orderkey | bigint | 12M | 5.38s | 6.69s | 19% |
| l_receiptdate | string, decimal, bigint | 2600 | 17.3s | 18.7s | 7% |
+---------------+-------------------------+------+----------+----------+-------------+
Change-Id: Ia4161a61db1782dc448dae9a1d4c1d120b055b3c
Reviewed-on: http://gerrit.cloudera.org:8080/18802
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/exec/partial-sort-node.cc | 4 +++-
be/src/exec/sort-node.cc | 4 +++-
be/src/exec/topn-node.cc | 3 ++-
be/src/runtime/sorter-internal.h | 5 ++++-
be/src/runtime/sorter-ir.cc | 6 +++---
be/src/runtime/sorter.cc | 6 +++++-
6 files changed, 20 insertions(+), 8 deletions(-)
diff --git a/be/src/exec/partial-sort-node.cc b/be/src/exec/partial-sort-node.cc
index 2ba5724ed..1fadbe523 100644
--- a/be/src/exec/partial-sort-node.cc
+++ b/be/src/exec/partial-sort-node.cc
@@ -94,7 +94,9 @@ void PartialSortPlanNode::Codegen(FragmentState* state) {
llvm::Function* compare_fn = nullptr;
AddCodegenStatus(row_comparator_config_->Codegen(state, &compare_fn));
AddCodegenStatus(
- Sorter::TupleSorter::Codegen(state, compare_fn, &codegend_sort_helper_fn_));
+ Sorter::TupleSorter::Codegen(state, compare_fn,
+ row_descriptor_->tuple_descriptors()[0]->byte_size(),
+ &codegend_sort_helper_fn_));
}
Status PartialSortNode::Open(RuntimeState* state) {
diff --git a/be/src/exec/sort-node.cc b/be/src/exec/sort-node.cc
index 739694c9e..223b689a9 100644
--- a/be/src/exec/sort-node.cc
+++ b/be/src/exec/sort-node.cc
@@ -111,7 +111,9 @@ void SortPlanNode::Codegen(FragmentState* state) {
Status codegen_status = row_comparator_config_->Codegen(state, &compare_fn);
if (codegen_status.ok()) {
codegen_status =
- Sorter::TupleSorter::Codegen(state, compare_fn, &codegend_sort_helper_fn_);
+ Sorter::TupleSorter::Codegen(state, compare_fn,
+ row_descriptor_->tuple_descriptors()[0]->byte_size(),
+ &codegend_sort_helper_fn_);
}
AddCodegenStatus(codegen_status);
}
diff --git a/be/src/exec/topn-node.cc b/be/src/exec/topn-node.cc
index f65af7df3..2e331a6a8 100644
--- a/be/src/exec/topn-node.cc
+++ b/be/src/exec/topn-node.cc
@@ -218,7 +218,8 @@ void TopNPlanNode::Codegen(FragmentState* state) {
Status codegen_status = ordering_comparator_config_->Codegen(state, &compare_fn);
if (codegen_status.ok() && is_partitioned()) {
codegen_status =
- Sorter::TupleSorter::Codegen(state, compare_fn, &codegend_sort_helper_fn_);
+ Sorter::TupleSorter::Codegen(state, compare_fn, output_tuple_desc_->byte_size(),
+ &codegend_sort_helper_fn_);
}
if (codegen_status.ok() && is_partitioned()) {
// TODO: IMPALA-10228: replace comparisons in std::map.
diff --git a/be/src/runtime/sorter-internal.h b/be/src/runtime/sorter-internal.h
index 5945526ba..931ee20b6 100644
--- a/be/src/runtime/sorter-internal.h
+++ b/be/src/runtime/sorter-internal.h
@@ -444,7 +444,7 @@ class Sorter::TupleSorter {
/// 'compare_fn' is the pointer to the code-gen version of the compare method with
/// which to replace all non-code-gen versions.
static Status Codegen(FragmentState* state, llvm::Function* compare_fn,
- CodegenFnPtr<SortHelperFn>* codegend_fn);
+ int tuple_byte_size, CodegenFnPtr<SortHelperFn>* codegend_fn);
/// Mangled name of SorterHelper().
static const char* SORTER_HELPER_SYMBOL;
@@ -460,6 +460,9 @@ class Sorter::TupleSorter {
/// Size of the tuples in memory.
const int tuple_size_;
+ /// Getter for the size of the tuples in memory. Replaced by a constant during codegen
+ int IR_NO_INLINE get_tuple_size() const { return tuple_size_; }
+
/// Tuple comparator with method Less() that returns true if lhs < rhs.
const TupleRowComparator& comparator_;
diff --git a/be/src/runtime/sorter-ir.cc b/be/src/runtime/sorter-ir.cc
index 16da09d61..44398fb5a 100644
--- a/be/src/runtime/sorter-ir.cc
+++ b/be/src/runtime/sorter-ir.cc
@@ -90,7 +90,7 @@ Status IR_ALWAYS_INLINE Sorter::TupleSorter::Partition2way(TupleIterator begin,
TupleIterator end, const Tuple* pivot, TupleIterator* cut) {
// Hoist member variable lookups out of loop to avoid extra loads inside loop.
Run* run = run_;
- int tuple_size = tuple_size_;
+ const int tuple_size = get_tuple_size();
Tuple* pivot_tuple = reinterpret_cast<Tuple*>(temp_tuple_buffer_);
TupleRow* pivot_tuple_row = reinterpret_cast<TupleRow*>(&pivot_tuple);
Tuple* swap_tuple = reinterpret_cast<Tuple*>(swap_buffer_);
@@ -136,7 +136,7 @@ Status IR_ALWAYS_INLINE Sorter::TupleSorter::Partition3way(TupleIterator begin,
TupleIterator* cut_right) {
// Hoist member variable lookups out of loop to avoid extra loads inside loop.
Run* run = run_;
- int tuple_size = tuple_size_;
+ const int tuple_size = get_tuple_size();
Tuple* pivot_tuple = reinterpret_cast<Tuple*>(temp_tuple_buffer_);
TupleRow* pivot_tuple_row = reinterpret_cast<TupleRow*>(&pivot_tuple);
Tuple* swap_tuple = reinterpret_cast<Tuple*>(swap_buffer_);
@@ -215,7 +215,7 @@ Status IR_ALWAYS_INLINE Sorter::TupleSorter::InsertionSort(const TupleIterator&
// Hoist member variable lookups out of loop to avoid extra loads inside loop.
Run* run = run_;
- int tuple_size = tuple_size_;
+ const int tuple_size = get_tuple_size();
uint8_t* temp_tuple_buffer = temp_tuple_buffer_;
TupleIterator insert_iter = begin;
diff --git a/be/src/runtime/sorter.cc b/be/src/runtime/sorter.cc
index 04fa338b0..5cb99d8e5 100644
--- a/be/src/runtime/sorter.cc
+++ b/be/src/runtime/sorter.cc
@@ -1211,7 +1211,7 @@ const char* Sorter::TupleSorter::LLVM_CLASS_NAME = "class.impala::Sorter::TupleS
// A method to code-gen for TupleSorter::SortHelper().
Status Sorter::TupleSorter::Codegen(FragmentState* state, llvm::Function* compare_fn,
- CodegenFnPtr<SortHelperFn>* codegened_fn) {
+ int tuple_size, CodegenFnPtr<SortHelperFn>* codegened_fn) {
LlvmCodeGen* codegen = state->codegen();
DCHECK(codegen != nullptr);
@@ -1228,6 +1228,10 @@ Status Sorter::TupleSorter::Codegen(FragmentState* state, llvm::Function* compar
replaced = codegen->ReplaceCallSites(fn, fn, SORTER_HELPER_SYMBOL);
DCHECK_REPLACE_COUNT(replaced, 2) << LlvmCodeGen::Print(fn);
+ replaced = codegen->ReplaceCallSitesWithValue(fn,
+ codegen->GetI32Constant(tuple_size), "get_tuple_size");
+ DCHECK_REPLACE_COUNT(replaced, 3) << LlvmCodeGen::Print(fn);
+
fn = codegen->FinalizeFunction(fn);
if (fn == nullptr) {
return Status("Sorter::TupleSorter::Codegen(): failed to finalize function");