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 2019/09/17 20:08:45 UTC
[impala] branch master updated: IMPALA-7322: Add storage wait time
to profile
This is an automated email from the ASF dual-hosted git repository.
stakiar pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
The following commit(s) were added to refs/heads/master by this push:
new 7136e8b IMPALA-7322: Add storage wait time to profile
7136e8b is described below
commit 7136e8b965bd0df974dccd1419ea65d42c494c06
Author: Yongzhi Chen <yc...@cloudera.com>
AuthorDate: Tue Jul 2 12:29:17 2019 -0400
IMPALA-7322: Add storage wait time to profile
Add metrics to record storage wait time for operations with
metadata load in catalog for hdfs, kudu and hbase tables.
Pass storage wait time from catalog to fe through thrift
and log total storage load time in query profile.
Storage-load-time is the amount of time spent loading metadata
from the underlying storage layer (e.g. S3, HDFS, Kudu, HBase),
which does not include the amount of time spending loading data
from HMS.
Testing:
* Ran queries that can trigger all of, none of or
some of the related tables loading.
* Check query profile for each query.
* Check catalog metrics for each table.
* Add unit tests to test_observability.py
* Ran all core tests.
Sample output:
Profile for Catalog V1: (storage-load-time is the added property and
it is part of Metadata load in Query Compilation):
After ran a hbase query (Metadata load finished is divided into
several lines because of limitation of commit message):
Query Compilation: 4s401ms
- Metadata load started: 661.084us (661.084us)
- Metadata load finished. loaded-tables=1/1
load-requests=1 catalog-updates=3
storage-load-time=233ms: 3s819ms (3s819ms)
- Analysis finished: 3s820ms (763.979us)
- Value transfer graph computed: 3s820ms (63.193us)
Profile for Catalog V2: (StorageLoad.Time is the added property and it
is in CatalogFetch):
Frontend:
- CatalogFetch.ColumnStats.Misses: 1
- CatalogFetch.ColumnStats.Requests: 1
- CatalogFetch.ColumnStats.Time: 0
- CatalogFetch.Config.Misses: 1
- CatalogFetch.Config.Requests: 1
- CatalogFetch.Config.Time: 3ms
- CatalogFetch.DatabaseList.Hits: 1
- CatalogFetch.DatabaseList.Requests: 1
- CatalogFetch.DatabaseList.Time: 0
- CatalogFetch.PartitionLists.Misses: 1
- CatalogFetch.PartitionLists.Requests: 1
- CatalogFetch.PartitionLists.Time: 4ms
- CatalogFetch.Partitions.Hits: 2
- CatalogFetch.Partitions.Misses: 1
- CatalogFetch.Partitions.Requests: 3
- CatalogFetch.Partitions.Time: 1ms
- CatalogFetch.RPCs.Bytes: 1.01 KB (1036)
- CatalogFetch.RPCs.Requests: 4
- CatalogFetch.RPCs.Time: 93ms
- CatalogFetch.StorageLoad.Time: 68ms
- CatalogFetch.TableNames.Hits: 2
- CatalogFetch.TableNames.Requests: 2
- CatalogFetch.TableNames.Time: 0
- CatalogFetch.Tables.Misses: 1
- CatalogFetch.Tables.Requests: 1
- CatalogFetch.Tables.Time: 91ms
Catalog metrics(this sample is from a hdfs table):
storage-metadata-load-duration:
Count: 1
Mean rate: 0.0085
1 min. rate: 0.032
5 min. rate: 0.1386
15 min. rate: 0.177
Min (msec): 111
Max (msec): 111
Mean (msec): 111.1802
Median (msec): 111.1802
75th-% (msec): 111.1802
95th-% (msec): 111.1802
99th-% (msec): 111.1802
Change-Id: I7447f8c8e7e50eb71d18643859d2e3de865368d2
Reviewed-on: http://gerrit.cloudera.org:8080/13786
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Reviewed-by: Sahil Takiar <st...@cloudera.com>
---
common/thrift/CatalogObjects.thrift | 4 ++
common/thrift/CatalogService.thrift | 4 ++
.../apache/impala/analysis/StmtMetadataLoader.java | 22 ++++++-
.../java/org/apache/impala/catalog/HBaseTable.java | 17 ++++--
.../java/org/apache/impala/catalog/HdfsTable.java | 67 ++++++++++++++++------
.../java/org/apache/impala/catalog/KuduTable.java | 9 ++-
.../main/java/org/apache/impala/catalog/Table.java | 14 +++++
.../impala/catalog/local/CatalogdMetaProvider.java | 20 ++++++-
tests/query_test/test_observability.py | 43 ++++++++++++++
9 files changed, 170 insertions(+), 30 deletions(-)
diff --git a/common/thrift/CatalogObjects.thrift b/common/thrift/CatalogObjects.thrift
index 01bba1d..b67d8d0 100644
--- a/common/thrift/CatalogObjects.thrift
+++ b/common/thrift/CatalogObjects.thrift
@@ -490,6 +490,10 @@ struct TTable {
// For example ValidReaderWriteIdList object's format is:
// <table_name>:<highwatermark>:<minOpenWriteId>:<open_writeids>:<abort_writeids>
14: optional string valid_write_ids
+
+ // Set if this table needs storage access during metadata load.
+ // Time used for storage loading in nanoseconds.
+ 15: optional i64 storage_metadata_load_time_ns
}
// Represents a database.
diff --git a/common/thrift/CatalogService.thrift b/common/thrift/CatalogService.thrift
index a290b96..205ee89 100644
--- a/common/thrift/CatalogService.thrift
+++ b/common/thrift/CatalogService.thrift
@@ -361,6 +361,10 @@ struct TPartialTableInfo {
3: optional list<hive_metastore.ColumnStatisticsObj> column_stats
+ // Set if this table needs storage access during metadata load.
+ // Time used for storage loading in nanoseconds.
+ 4: optional i64 storage_metadata_load_time_ns
+
// Each TNetworkAddress is a datanode which contains blocks of a file in the table.
// Used so that each THdfsFileBlock can just reference an index in this list rather
// than duplicate the list of network address, which helps reduce memory usage.
diff --git a/fe/src/main/java/org/apache/impala/analysis/StmtMetadataLoader.java b/fe/src/main/java/org/apache/impala/analysis/StmtMetadataLoader.java
index d976fad..42f9ae4 100644
--- a/fe/src/main/java/org/apache/impala/analysis/StmtMetadataLoader.java
+++ b/fe/src/main/java/org/apache/impala/analysis/StmtMetadataLoader.java
@@ -18,6 +18,7 @@
package org.apache.impala.analysis;
import java.util.ArrayList;
+import java.util.concurrent.TimeUnit;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -29,6 +30,7 @@ import org.apache.impala.catalog.FeDb;
import org.apache.impala.catalog.FeIncompleteTable;
import org.apache.impala.catalog.FeTable;
import org.apache.impala.catalog.FeView;
+import org.apache.impala.catalog.Table;
import org.apache.impala.common.InternalException;
import org.apache.impala.compat.MetastoreShim;
import org.apache.impala.service.Frontend;
@@ -226,11 +228,25 @@ public class StmtMetadataLoader {
missingTbls = newMissingTbls;
++numCatalogUpdatesReceived_;
}
+
if (timeline_ != null) {
- timeline_.markEvent(String.format("Metadata load finished. " +
- "loaded-tables=%d/%d load-requests=%d catalog-updates=%d",
+ long storageLoadTimeNano = 0;
+ // Calculate the total storage loading time for this query (not including
+ // the tables already loaded before the query was called).
+ storageLoadTimeNano =
+ loadedOrFailedTbls_.values()
+ .stream()
+ .filter(Table.class::isInstance)
+ .map(Table.class::cast)
+ .filter(loadedTbl -> requestedTbls.contains(loadedTbl.getTableName()))
+ .mapToLong(Table::getStorageLoadTime)
+ .sum();
+ timeline_.markEvent(String.format("Metadata load finished. "
+ + "loaded-tables=%d/%d load-requests=%d catalog-updates=%d "
+ + "storage-load-time=%dms",
requestedTbls.size(), loadedOrFailedTbls_.size(), numLoadRequestsSent_,
- numCatalogUpdatesReceived_));
+ numCatalogUpdatesReceived_,
+ TimeUnit.MILLISECONDS.convert(storageLoadTimeNano, TimeUnit.NANOSECONDS)));
if (MetastoreShim.getMajorVersion() > 2) {
StringBuilder validIdsBuf = new StringBuilder("Loaded ValidWriteIdLists");
diff --git a/fe/src/main/java/org/apache/impala/catalog/HBaseTable.java b/fe/src/main/java/org/apache/impala/catalog/HBaseTable.java
index 5e7280f..9e62459 100644
--- a/fe/src/main/java/org/apache/impala/catalog/HBaseTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/HBaseTable.java
@@ -103,11 +103,18 @@ public class HBaseTable extends Table implements FeHBaseTable {
Preconditions.checkNotNull(getMetaStoreTable());
try (Timer.Context timer = getMetrics().getTimer(Table.LOAD_DURATION_METRIC).time()) {
msTable_ = msTbl;
- hbaseTableName_ = Util.getHBaseTableName(getMetaStoreTable());
- // Warm up the connection and verify the table exists.
- Util.getHBaseTable(hbaseTableName_).close();
- columnFamilies_ = null;
- List<Column> cols = Util.loadColumns(msTable_);
+ final Timer.Context storageLoadTimer =
+ getMetrics().getTimer(Table.STORAGE_METADATA_LOAD_DURATION_METRIC).time();
+ List<Column> cols;
+ try {
+ hbaseTableName_ = Util.getHBaseTableName(getMetaStoreTable());
+ // Warm up the connection and verify the table exists.
+ Util.getHBaseTable(hbaseTableName_).close();
+ columnFamilies_ = null;
+ cols = Util.loadColumns(msTable_);
+ } finally {
+ storageMetadataLoadTime_ = storageLoadTimer.stop();
+ }
clearColumns();
for (Column col : cols) addColumn(col);
// Set table stats.
diff --git a/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java b/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
index 8f440f2..d2039f1 100644
--- a/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
@@ -31,6 +31,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
+import java.util.concurrent.TimeUnit;
import org.apache.avro.Schema;
import org.apache.hadoop.conf.Configuration;
@@ -87,6 +88,7 @@ import org.apache.impala.util.ThreadNameAnnotator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.codahale.metrics.Clock;
import com.codahale.metrics.Gauge;
import com.codahale.metrics.Timer;
import com.google.common.annotations.VisibleForTesting;
@@ -504,16 +506,18 @@ public class HdfsTable extends Table implements FeFsTable {
* Create HdfsPartition objects corresponding to 'msPartitions' and add them to this
* table's partition list. Any partition metadata will be reset and loaded from
* scratch. For each partition created, we load the block metadata for each data file
- * under it.
+ * under it. Returns time spent loading the filesystem metadata in nanoseconds.
*
* If there are no partitions in the Hive metadata, a single partition is added with no
* partition keys.
*/
- private void loadAllPartitions(
+ private long loadAllPartitions(
List<org.apache.hadoop.hive.metastore.api.Partition> msPartitions,
org.apache.hadoop.hive.metastore.api.Table msTbl) throws IOException,
CatalogException {
Preconditions.checkNotNull(msTbl);
+ final Clock clock = Clock.defaultClock();
+ long startTime = clock.getTick();
initializePartitionMetadata(msTbl);
FsPermissionCache permCache = preloadPermissionsCache(msPartitions);
@@ -541,6 +545,7 @@ public class HdfsTable extends Table implements FeFsTable {
}
// Load the file metadata from scratch.
loadFileMetadataForPartitions(partitionMap_.values(), /*isRefresh=*/false);
+ return clock.getTick() - startTime;
}
@@ -926,7 +931,10 @@ public class HdfsTable extends Table implements FeFsTable {
loadTableSchema ? "table definition and " : "",
partitionsToUpdate == null ? "all" : String.valueOf(partitionsToUpdate.size()),
msTbl.getDbName(), msTbl.getTableName(), reason);
- LOG.info(annotation);;
+ LOG.info(annotation);
+ final Timer storageLdTimer =
+ getMetrics().getTimer(Table.STORAGE_METADATA_LOAD_DURATION_METRIC);
+ storageMetadataLoadTime_ = 0;
try (ThreadNameAnnotator tna = new ThreadNameAnnotator(annotation)) {
// turn all exceptions into TableLoadingException
msTable_ = msTbl;
@@ -939,16 +947,19 @@ public class HdfsTable extends Table implements FeFsTable {
loadAllColumnStats(client);
}
loadValidWriteIdList(client);
+
// Load partition and file metadata
if (reuseMetadata) {
// Incrementally update this table's partitions and file metadata
Preconditions.checkState(
partitionsToUpdate == null || loadParitionFileMetadata);
- updateMdFromHmsTable(msTbl);
+ storageMetadataLoadTime_ += updateMdFromHmsTable(msTbl);
if (msTbl.getPartitionKeysSize() == 0) {
- if (loadParitionFileMetadata) updateUnpartitionedTableFileMd();
+ if (loadParitionFileMetadata) {
+ storageMetadataLoadTime_ += updateUnpartitionedTableFileMd();
+ }
} else {
- updatePartitionsFromHms(
+ storageMetadataLoadTime_ += updatePartitionsFromHms(
client, partitionsToUpdate, loadParitionFileMetadata);
}
LOG.info("Incrementally loaded table metadata for: " + getFullName());
@@ -959,7 +970,7 @@ public class HdfsTable extends Table implements FeFsTable {
MetaStoreUtil.fetchAllPartitions(
client, db_.getName(), name_, NUM_PARTITION_FETCH_RETRIES);
LOG.info("Fetched partition metadata from the Metastore: " + getFullName());
- loadAllPartitions(msPartitions, msTbl);
+ storageMetadataLoadTime_ = loadAllPartitions(msPartitions, msTbl);
}
if (loadTableSchema) setAvroSchema(client, msTbl);
setTableStats(msTbl);
@@ -972,37 +983,47 @@ public class HdfsTable extends Table implements FeFsTable {
+ getFullName(), e);
}
} finally {
+ storageLdTimer.update(storageMetadataLoadTime_, TimeUnit.NANOSECONDS);
context.stop();
}
}
/**
* Updates the table metadata, including 'hdfsBaseDir_', 'isMarkedCached_',
- * and 'accessLevel_' from 'msTbl'. Throws an IOException if there was an error
- * accessing the table location path.
+ * and 'accessLevel_' from 'msTbl'. Returns time spent accessing file system
+ * in nanoseconds. Throws an IOException if there was an error accessing
+ * the table location path.
*/
- private void updateMdFromHmsTable(org.apache.hadoop.hive.metastore.api.Table msTbl)
+ private long updateMdFromHmsTable(org.apache.hadoop.hive.metastore.api.Table msTbl)
throws IOException {
Preconditions.checkNotNull(msTbl);
+ final Clock clock = Clock.defaultClock();
+ long filesystemAccessTime = 0;
+ long startTime = clock.getTick();
hdfsBaseDir_ = msTbl.getSd().getLocation();
isMarkedCached_ = HdfsCachingUtil.validateCacheParams(msTbl.getParameters());
Path location = new Path(hdfsBaseDir_);
accessLevel_ = getAvailableAccessLevel(getFullName(), location,
new FsPermissionCache());
+ filesystemAccessTime = clock.getTick() - startTime;
setMetaStoreTable(msTbl);
+ return filesystemAccessTime;
}
/**
* Incrementally updates the file metadata of an unpartitioned HdfsTable.
+ * Returns time spent updating the file metadata in nanoseconds.
*
* This is optimized for the case where few files have changed. See
* {@link #refreshFileMetadata(Path, List)} above for details.
*/
- private void updateUnpartitionedTableFileMd() throws CatalogException {
+ private long updateUnpartitionedTableFileMd() throws CatalogException {
Preconditions.checkState(getNumClusteringCols() == 0);
if (LOG.isTraceEnabled()) {
LOG.trace("update unpartitioned table: " + getFullName());
}
+ final Clock clock = Clock.defaultClock();
+ long startTime = clock.getTick();
HdfsPartition oldPartition = Iterables.getOnlyElement(partitionMap_.values());
// Instead of updating the existing partition in place, we create a new one
@@ -1020,18 +1041,21 @@ public class HdfsTable extends Table implements FeFsTable {
addPartition(part);
if (isMarkedCached_) part.markCached();
loadFileMetadataForPartitions(ImmutableList.of(part), /*isRefresh=*/true);
+ return clock.getTick() - startTime;
}
/**
- * Updates the partitions of an HdfsTable so that they are in sync with the Hive
- * Metastore. It reloads partitions that were marked 'dirty' by doing a DROP + CREATE.
- * It removes from this table partitions that no longer exist in the Hive Metastore and
- * adds partitions that were added externally (e.g. using Hive) to the Hive Metastore
- * but do not exist in this table. If 'loadParitionFileMetadata' is true, it triggers
- * file/block metadata reload for the partitions specified in 'partitionsToUpdate', if
- * any, or for all the table partitions if 'partitionsToUpdate' is null.
+ * Updates the partitions of an HdfsTable so that they are in sync with the
+ * Hive Metastore. It reloads partitions that were marked 'dirty' by doing a
+ * DROP + CREATE. It removes from this table partitions that no longer exist
+ * in the Hive Metastore and adds partitions that were added externally (e.g.
+ * using Hive) to the Hive Metastore but do not exist in this table. If
+ * 'loadParitionFileMetadata' is true, it triggers file/block metadata reload
+ * for the partitions specified in 'partitionsToUpdate', if any, or for all
+ * the table partitions if 'partitionsToUpdate' is null. Returns time
+ * spent loading file metadata in nanoseconds.
*/
- private void updatePartitionsFromHms(IMetaStoreClient client,
+ private long updatePartitionsFromHms(IMetaStoreClient client,
Set<String> partitionsToUpdate, boolean loadPartitionFileMetadata)
throws Exception {
if (LOG.isTraceEnabled()) LOG.trace("Sync table partitions: " + getFullName());
@@ -1095,14 +1119,19 @@ public class HdfsTable extends Table implements FeFsTable {
// Load file metadata. Until we have a notification mechanism for when a
// file changes in hdfs, it is sometimes required to reload all the file
// descriptors and block metadata of a table (e.g. REFRESH statement).
+ long fileLoadMdTime = 0;
if (loadPartitionFileMetadata) {
+ final Clock clock = Clock.defaultClock();
+ long startTime = clock.getTick();
if (partitionsToUpdate != null) {
Preconditions.checkState(partitionsToLoadFiles.isEmpty());
// Only reload file metadata of partitions specified in 'partitionsToUpdate'
partitionsToLoadFiles = getPartitionsForNames(partitionsToUpdate);
}
loadFileMetadataForPartitions(partitionsToLoadFiles, /* isRefresh=*/true);
+ fileLoadMdTime = clock.getTick() - startTime;
}
+ return fileLoadMdTime;
}
/**
diff --git a/fe/src/main/java/org/apache/impala/catalog/KuduTable.java b/fe/src/main/java/org/apache/impala/catalog/KuduTable.java
index 3db60ee..058b09d 100644
--- a/fe/src/main/java/org/apache/impala/catalog/KuduTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/KuduTable.java
@@ -285,14 +285,19 @@ public class KuduTable extends Table implements FeKuduTable {
" property found for Kudu table " + kuduTableName_);
}
setTableStats(msTable_);
- // Load metadata from Kudu and HMS
+ // Load metadata from Kudu
+ final Timer.Context ctxStorageLdTime =
+ getMetrics().getTimer(Table.STORAGE_METADATA_LOAD_DURATION_METRIC).time();
try {
loadSchemaFromKudu();
- loadAllColumnStats(msClient);
} catch (ImpalaRuntimeException e) {
throw new TableLoadingException("Error loading metadata for Kudu table " +
kuduTableName_, e);
+ } finally {
+ storageMetadataLoadTime_ = ctxStorageLdTime.stop();
}
+ // Load from HMS
+ loadAllColumnStats(msClient);
refreshLastUsedTime();
// Avoid updating HMS if the schema didn't change.
if (msTable_.equals(msTbl)) return;
diff --git a/fe/src/main/java/org/apache/impala/catalog/Table.java b/fe/src/main/java/org/apache/impala/catalog/Table.java
index 7683381..e3adac3 100644
--- a/fe/src/main/java/org/apache/impala/catalog/Table.java
+++ b/fe/src/main/java/org/apache/impala/catalog/Table.java
@@ -115,6 +115,9 @@ public abstract class Table extends CatalogObjectImpl implements FeTable {
// True if this object is stored in an Impalad catalog cache.
protected boolean storedInImpaladCatalogCache_ = false;
+ // Time spent in the source systems loading/reloading the fs metadata for the table.
+ protected long storageMetadataLoadTime_ = 0;
+
// Last used time of this table in nanoseconds as returned by
// CatalogdTableInvalidator.nanoTime(). This is only set in catalogd and not used by
// impalad.
@@ -138,6 +141,8 @@ public abstract class Table extends CatalogObjectImpl implements FeTable {
public static final String REFRESH_DURATION_METRIC = "refresh-duration";
public static final String ALTER_DURATION_METRIC = "alter-duration";
public static final String LOAD_DURATION_METRIC = "load-duration";
+ public static final String STORAGE_METADATA_LOAD_DURATION_METRIC =
+ "storage-metadata-load-duration";
// Table property key for storing the time of the last DDL operation.
public static final String TBL_PROP_LAST_DDL_TIME = "transient_lastDdlTime";
@@ -197,10 +202,14 @@ public abstract class Table extends CatalogObjectImpl implements FeTable {
metrics_.addTimer(REFRESH_DURATION_METRIC);
metrics_.addTimer(ALTER_DURATION_METRIC);
metrics_.addTimer(LOAD_DURATION_METRIC);
+ metrics_.addTimer(STORAGE_METADATA_LOAD_DURATION_METRIC);
}
public Metrics getMetrics() { return metrics_; }
+ // Returns storage wait time during metadata load.
+ public long getStorageLoadTime() { return storageMetadataLoadTime_; }
+
// Returns true if this table reference comes from the impalad catalog cache or if it
// is loaded from the testing framework. Returns false if this table reference points
// to a table stored in the catalog server.
@@ -400,6 +409,8 @@ public abstract class Table extends CatalogObjectImpl implements FeTable {
accessLevel_ = thriftTable.isSetAccess_level() ? thriftTable.getAccess_level() :
TAccessLevel.READ_WRITE;
+ storageMetadataLoadTime_ = thriftTable.getStorage_metadata_load_time_ns();
+
storedInImpaladCatalogCache_ = true;
validWriteIds_ = thriftTable.isSetValid_write_ids() ?
thriftTable.getValid_write_ids() : null;
@@ -436,6 +447,7 @@ public abstract class Table extends CatalogObjectImpl implements FeTable {
TTable table = new TTable(db_.getName(), name_);
table.setAccess_level(accessLevel_);
+ table.setStorage_metadata_load_time_ns(storageMetadataLoadTime_);
// Populate both regular columns and clustering columns (if there are any).
table.setColumns(new ArrayList<>());
@@ -485,6 +497,8 @@ public abstract class Table extends CatalogObjectImpl implements FeTable {
TGetPartialCatalogObjectResponse resp = new TGetPartialCatalogObjectResponse();
resp.setObject_version_number(getCatalogVersion());
resp.table_info = new TPartialTableInfo();
+ resp.table_info.setStorage_metadata_load_time_ns(storageMetadataLoadTime_);
+ storageMetadataLoadTime_ = 0;
if (selector.want_hms_table) {
// TODO(todd): the deep copy could be a bit expensive. Unfortunately if we took
// a reference to this object, and let it escape out of the lock, it would be
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 81432de..da88faf 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
@@ -209,6 +209,7 @@ public class CatalogdMetaProvider implements MetaProvider {
private static final String FUNCTION_LIST_STATS_CATEGORY = "FunctionLists";
private static final String FUNCTIONS_STATS_CATEGORY = "Functions";
private static final String RPC_STATS_CATEGORY = "RPCs";
+ private static final String STORAGE_METADATA_LOAD_CATEGORY = "StorageLoad";
private static final String RPC_REQUESTS =
CATALOG_FETCH_PREFIX + "." + RPC_STATS_CATEGORY + ".Requests";
private static final String RPC_BYTES =
@@ -543,6 +544,20 @@ public class CatalogdMetaProvider implements MetaProvider {
}
}
+ /**
+ * Adds tables metadata storage access time to query's profile.
+ * The access time is aggregated for the tables which need to be loaded.
+ */
+ private void addTableMetadatStorageLoadTimeToProfile(long storageLoadTimeNano) {
+ FrontendProfile profile = FrontendProfile.getCurrentOrNull();
+ if (profile == null) return;
+ // Storage-load-time for the table and its partitions
+ final String storageAccessTimeCounter = CATALOG_FETCH_PREFIX + "." +
+ STORAGE_METADATA_LOAD_CATEGORY + "." + "Time";
+ profile.addToCounter(storageAccessTimeCounter, TUnit.TIME_MS,
+ TimeUnit.MILLISECONDS.convert(storageLoadTimeNano, TimeUnit.NANOSECONDS));
+ }
+
@Override
public ImmutableList<String> loadDbList() throws TException {
return loadWithCaching("database list", DB_LIST_STATS_CATEGORY, DB_LIST_CACHE_KEY,
@@ -663,6 +678,8 @@ public class CatalogdMetaProvider implements MetaProvider {
TGetPartialCatalogObjectResponse resp = sendRequest(req);
checkResponse(resp.table_info != null && resp.table_info.hms_table != null,
req, "missing expected HMS table");
+ addTableMetadatStorageLoadTimeToProfile(
+ resp.table_info.storage_metadata_load_time_ns);
return new TableMetaRefImpl(
dbName, tableName, resp.table_info.hms_table, resp.object_version_number);
}
@@ -830,7 +847,8 @@ public class CatalogdMetaProvider implements MetaProvider {
checkResponse(resp.table_info.partitions.size() == ids.size(),
req, "returned %d partitions instead of expected %d",
resp.table_info.partitions.size(), ids.size());
-
+ addTableMetadatStorageLoadTimeToProfile(
+ resp.table_info.storage_metadata_load_time_ns);
Map<PartitionRef, PartitionMetadata> ret = new HashMap<>();
for (int i = 0; i < ids.size(); i++) {
PartitionRef partRef = partRefs.get(i);
diff --git a/tests/query_test/test_observability.py b/tests/query_test/test_observability.py
index fe2d92b..09674fd 100644
--- a/tests/query_test/test_observability.py
+++ b/tests/query_test/test_observability.py
@@ -331,6 +331,7 @@ class TestObservability(ImpalaTestSuite):
r'CatalogFetch.RPCs.Bytes',
r'CatalogFetch.RPCs.Requests',
r'CatalogFetch.RPCs.Time',
+ r'CatalogFetch.StorageLoad.Time',
r'CatalogFetch.TableNames.Hits',
r'CatalogFetch.TableNames.Requests',
r'CatalogFetch.TableNames.Time',
@@ -666,3 +667,45 @@ class TestObservability(ImpalaTestSuite):
query = "select count (*) from functional.alltypes"
runtime_profile = self.execute_query(query).runtime_profile
assert "Executor Group:" in runtime_profile
+
+ def test_query_profile_storage_load_time_filesystem(self, unique_database,
+ cluster_properties):
+ """Test that when a query needs load metadata for table(s), the
+ storage load time should be in the profile. Tests file systems."""
+ table_name = 'ld_prof'
+ self.execute_query(
+ "create table {0}.{1}(col1 int)".format(unique_database, table_name))
+ self.__check_query_profile_storage_load_time(unique_database, table_name,
+ cluster_properties)
+
+ @SkipIfS3.hbase
+ @SkipIfLocal.hbase
+ @SkipIfIsilon.hbase
+ @SkipIfABFS.hbase
+ @SkipIfADLS.hbase
+ @pytest.mark.execute_serially
+ def test_query_profile_storage_load_time(self, cluster_properties):
+ """Test that when a query needs load metadata for table(s), the
+ storage load time should be in the profile. Tests kudu and hbase."""
+ # KUDU table
+ self.__check_query_profile_storage_load_time("functional_kudu", "alltypes",
+ cluster_properties)
+
+ # HBASE table
+ self.__check_query_profile_storage_load_time("functional_hbase", "alltypes",
+ cluster_properties)
+
+ def __check_query_profile_storage_load_time(self, db_name, table_name,
+ cluster_properties):
+ """Check query profile for storage load time with a given database."""
+ self.execute_query("invalidate metadata {0}.{1}".format(db_name, table_name))
+ query = "select count (*) from {0}.{1}".format(db_name, table_name)
+ runtime_profile = self.execute_query(query).runtime_profile
+ if cluster_properties.is_catalog_v2_cluster():
+ storageLoadTime = "StorageLoad.Time"
+ else:
+ storageLoadTime = "storage-load-time"
+ assert storageLoadTime in runtime_profile
+ # Call the second time, no metastore loading needed.
+ runtime_profile = self.execute_query(query).runtime_profile
+ assert storageLoadTime not in runtime_profile