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 2023/12/01 07:22:04 UTC
(impala) branch master updated: IMPALA-12486: Add catalog metrics for metadata loading
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
The following commit(s) were added to refs/heads/master by this push:
new 9011b81af IMPALA-12486: Add catalog metrics for metadata loading
9011b81af is described below
commit 9011b81afa33ef7e4b0ec8a367b2713be8917213
Author: stiga-huang <hu...@gmail.com>
AuthorDate: Mon Nov 6 21:42:40 2023 +0800
IMPALA-12486: Add catalog metrics for metadata loading
This patch adds the following catalog metrics which indicate the load on
HDFS for loading file metadata:
- catalog-server.metadata.file.num-loading-threads: The total size of
all thread pools used in loading file metadata.
- catalog-server.metadata.file.num-loading-tasks: The total number of
unfinished file metadata loading tasks. Each task corresponds to a
partition.
- catalog-server.metadata.table.num-loading-file-metadata: The total
number of tables that are loading file metadata.
Also adds some metrics for metadata loading on all tables. Note that
metadata loading of an HDFS table consists of loading HMS metadata and
HDFS file metadata, etc.
- catalog-server.metadata.table.num-loading-metadata: The total number
of tables that are loading metadata.
- catalog-server.metadata.table.async-loading.num-in-progress: The
total number of tables that are loading metadata asynchronously. E.g.
the initial metadata loading triggered by the first access on a
table.
- catalog-server.metadata.table.async-loading.queue-len: The total
number of tables that are waiting for asynchronous loading. If this
number raises, consider bumping --num_metadata_loading_threads.
Three metrics about the catalog cache are also added:
- catalog.num-databases
- catalog.num-tables
- catalog.num-functions
Note that the first two are also shown in WebUI of coordinators and we
plan to deprecate them and only show them in catalogd's WebUI.
The number of idle and in-use HMS clients is also exposed in this
patch:
- catalog.hms-client-pool.num-idle
- catalog.hms-client-pool.num-in-use
Tests
- Launch catalogd locally with load_catalog_in_background=true and
verified the metrics.
- Add e2e tests in tests/webserver/test_web_pages.py
Change-Id: Icef7b123bdcb0f5b8572635eeaacd8294990f9ba
Reviewed-on: http://gerrit.cloudera.org:8080/20673
Reviewed-by: Andrew Sherman <as...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/catalog/catalog-server.cc | 52 ++++++++++++
be/src/catalog/catalog-server.h | 30 +++++++
common/thrift/JniCatalog.thrift | 17 ++++
common/thrift/metrics.json | 96 +++++++++++++++++++++-
.../java/org/apache/impala/catalog/Catalog.java | 8 ++
.../apache/impala/catalog/CatalogObjectCache.java | 2 +
.../impala/catalog/CatalogServiceCatalog.java | 29 +++++++
.../org/apache/impala/catalog/DataSourceTable.java | 4 +
fe/src/main/java/org/apache/impala/catalog/Db.java | 5 ++
.../apache/impala/catalog/FileMetadataLoader.java | 14 ++++
.../java/org/apache/impala/catalog/HBaseTable.java | 3 +
.../java/org/apache/impala/catalog/HdfsTable.java | 2 +
.../impala/catalog/IcebergFileMetadataLoader.java | 22 ++++-
.../java/org/apache/impala/catalog/KuduTable.java | 2 +
.../apache/impala/catalog/MetaStoreClientPool.java | 9 ++
.../impala/catalog/ParallelFileMetadataLoader.java | 38 ++++++---
.../main/java/org/apache/impala/catalog/Table.java | 3 +
.../org/apache/impala/catalog/TableLoadingMgr.java | 8 ++
.../main/java/org/apache/impala/catalog/View.java | 3 +
.../java/org/apache/impala/service/JniCatalog.java | 19 +++++
tests/webserver/test_web_pages.py | 17 ++++
21 files changed, 367 insertions(+), 16 deletions(-)
diff --git a/be/src/catalog/catalog-server.cc b/be/src/catalog/catalog-server.cc
index 1d2d678f1..6544ba6b2 100644
--- a/be/src/catalog/catalog-server.cc
+++ b/be/src/catalog/catalog-server.cc
@@ -180,6 +180,23 @@ const string CATALOG_SERVER_PARTIAL_FETCH_RPC_QUEUE_LEN =
const string CATALOG_ACTIVE_STATUS = "catalog-server.active-status";
const string CATALOG_HA_NUM_ACTIVE_STATUS_CHANGE =
"catalog-server.ha-number-active-status-change";
+const string CATALOG_NUM_FILE_METADATA_LOADING_THREADS =
+ "catalog-server.metadata.file.num-loading-threads";
+const string CATALOG_NUM_FILE_METADATA_LOADING_TASKS =
+ "catalog-server.metadata.file.num-loading-tasks";
+const string CATALOG_NUM_TABLES_LOADING_FILE_METADATA =
+ "catalog-server.metadata.table.num-loading-file-metadata";
+const string CATALOG_NUM_TABLES_LOADING_METADATA =
+ "catalog-server.metadata.table.num-loading-metadata";
+const string CATALOG_NUM_TABLES_ASYNC_LOADING_METADATA =
+ "catalog-server.metadata.table.async-loading.num-in-progress";
+const string CATALOG_NUM_TABLES_WAITING_FOR_ASYNC_LOADING =
+ "catalog-server.metadata.table.async-loading.queue-len";
+const string CATALOG_NUM_DBS = "catalog.num-databases";
+const string CATALOG_NUM_TABLES = "catalog.num-tables";
+const string CATALOG_NUM_FUNCTIONS = "catalog.num-functions";
+const string CATALOG_NUM_HMS_CLIENTS_IDLE = "catalog.hms-client-pool.num-idle";
+const string CATALOG_NUM_HMS_CLIENTS_IN_USE = "catalog.hms-client-pool.num-in-use";
const string CATALOG_WEB_PAGE = "/catalog";
const string CATALOG_TEMPLATE = "catalog.tmpl";
@@ -370,6 +387,24 @@ CatalogServer::CatalogServer(MetricGroup* metrics)
CATALOG_SERVER_TOPIC_PROCESSING_TIMES);
partial_fetch_rpc_queue_len_metric_ =
metrics->AddGauge(CATALOG_SERVER_PARTIAL_FETCH_RPC_QUEUE_LEN, 0);
+ num_file_metadata_loading_threads_metric_ =
+ metrics->AddGauge(CATALOG_NUM_FILE_METADATA_LOADING_THREADS, 0);
+ num_file_metadata_loading_tasks_metric_ =
+ metrics->AddGauge(CATALOG_NUM_FILE_METADATA_LOADING_TASKS, 0);
+ num_tables_loading_file_metadata_metric_ =
+ metrics->AddGauge(CATALOG_NUM_TABLES_LOADING_FILE_METADATA, 0);
+ num_tables_loading_metadata_metric_ =
+ metrics->AddGauge(CATALOG_NUM_TABLES_LOADING_METADATA, 0);
+ num_tables_async_loading_metadata_metric_ =
+ metrics->AddGauge(CATALOG_NUM_TABLES_ASYNC_LOADING_METADATA, 0);
+ num_tables_waiting_for_async_loading_metric_ =
+ metrics->AddGauge(CATALOG_NUM_TABLES_WAITING_FOR_ASYNC_LOADING, 0);
+ num_dbs_metric_ = metrics->AddGauge(CATALOG_NUM_DBS, 0);
+ num_tables_metric_ = metrics->AddGauge(CATALOG_NUM_TABLES, 0);
+ num_functions_metric_ = metrics->AddGauge(CATALOG_NUM_FUNCTIONS, 0);
+ num_hms_clients_idle_metric_ = metrics->AddGauge(CATALOG_NUM_HMS_CLIENTS_IDLE, 0);
+ num_hms_clients_in_use_metric_ = metrics->AddGauge(CATALOG_NUM_HMS_CLIENTS_IN_USE, 0);
+
active_status_metric_ =
metrics->AddProperty(CATALOG_ACTIVE_STATUS, !FLAGS_enable_catalogd_ha);
num_ha_active_status_change_metric_ =
@@ -632,6 +667,23 @@ bool CatalogServer::IsActive() {
}
partial_fetch_rpc_queue_len_metric_->SetValue(
response.catalog_partial_fetch_rpc_queue_len);
+ num_file_metadata_loading_threads_metric_->SetValue(
+ response.catalog_num_file_metadata_loading_threads);
+ num_file_metadata_loading_tasks_metric_->SetValue(
+ response.catalog_num_file_metadata_loading_tasks);
+ num_tables_loading_file_metadata_metric_->SetValue(
+ response.catalog_num_tables_loading_file_metadata);
+ num_tables_loading_metadata_metric_->SetValue(
+ response.catalog_num_tables_loading_metadata);
+ num_tables_async_loading_metadata_metric_->SetValue(
+ response.catalog_num_tables_async_loading_metadata);
+ num_tables_waiting_for_async_loading_metric_->SetValue(
+ response.catalog_num_tables_waiting_for_async_loading);
+ num_dbs_metric_->SetValue(response.catalog_num_dbs);
+ num_tables_metric_->SetValue(response.catalog_num_tables);
+ num_functions_metric_->SetValue(response.catalog_num_functions);
+ num_hms_clients_idle_metric_->SetValue(response.catalog_num_hms_clients_idle);
+ num_hms_clients_in_use_metric_->SetValue(response.catalog_num_hms_clients_in_use);
TEventProcessorMetrics eventProcessorMetrics = response.event_metrics;
MetastoreEventMetrics::refresh(&eventProcessorMetrics);
}
diff --git a/be/src/catalog/catalog-server.h b/be/src/catalog/catalog-server.h
index 6d1dde18a..e8c4513db 100644
--- a/be/src/catalog/catalog-server.h
+++ b/be/src/catalog/catalog-server.h
@@ -154,6 +154,36 @@ class CatalogServer {
/// Metric to count the number of active status changes.
IntCounter* num_ha_active_status_change_metric_;
+ /// Metric that tracks the total size of all thread pools used in all
+ /// ParallelFileMetadataLoader instances.
+ IntGauge* num_file_metadata_loading_threads_metric_;
+
+ /// Metric that tracks the total number of unfinished FileMetadataLoader tasks that
+ /// are submitted to the pools.
+ IntGauge* num_file_metadata_loading_tasks_metric_;
+
+ /// Metric that tracks the total number of tables that are loading file metadata.
+ IntGauge* num_tables_loading_file_metadata_metric_;
+
+ /// Metric that tracks the total number of tables that are loading metadata.
+ IntGauge* num_tables_loading_metadata_metric_;
+
+ /// Metric that tracks the total number of tables that are loading metadata
+ /// asynchronously, e.g. initial metadata loading triggered by first access of a table.
+ IntGauge* num_tables_async_loading_metadata_metric_;
+
+ /// Metric that tracks the total number of tables that are waiting for async loading.
+ IntGauge* num_tables_waiting_for_async_loading_metric_;
+
+ /// Metrics of the total number of dbs, tables and functions in the catalog cache
+ IntGauge* num_dbs_metric_;
+ IntGauge* num_tables_metric_;
+ IntGauge* num_functions_metric_;
+
+ /// Metrics that track the number of HMS clients
+ IntGauge* num_hms_clients_idle_metric_;
+ IntGauge* num_hms_clients_in_use_metric_;
+
/// Thread that polls the catalog for any updates.
std::unique_ptr<Thread> catalog_update_gathering_thread_;
diff --git a/common/thrift/JniCatalog.thrift b/common/thrift/JniCatalog.thrift
index 5297cc4dd..879bb610a 100755
--- a/common/thrift/JniCatalog.thrift
+++ b/common/thrift/JniCatalog.thrift
@@ -1038,6 +1038,23 @@ struct TGetCatalogServerMetricsResponse {
// get the catalogd Hive metastore server metrics, if configured
3: optional TCatalogdHmsCacheMetrics catalogd_hms_cache_metrics
+
+ // Metrics of table metadata loading
+ 4: optional i32 catalog_num_file_metadata_loading_threads
+ 5: optional i32 catalog_num_file_metadata_loading_tasks
+ 6: optional i32 catalog_num_tables_loading_file_metadata
+ 7: optional i32 catalog_num_tables_loading_metadata
+ 8: optional i32 catalog_num_tables_async_loading_metadata
+ 9: optional i32 catalog_num_tables_waiting_for_async_loading
+
+ // Metrics of the catalog
+ 10: optional i32 catalog_num_dbs
+ 11: optional i32 catalog_num_tables
+ 12: optional i32 catalog_num_functions
+
+ // Metrics of HMS clients
+ 13: optional i32 catalog_num_hms_clients_idle
+ 14: optional i32 catalog_num_hms_clients_in_use
}
// Request to copy the generated testcase from a given input path.
diff --git a/common/thrift/metrics.json b/common/thrift/metrics.json
index 55de02f44..78622bb90 100644
--- a/common/thrift/metrics.json
+++ b/common/thrift/metrics.json
@@ -330,8 +330,69 @@
"key": "catalog-server.topic-processing-time-s"
},
{
- "description": "The number of databases in the catalog. Untracked in LocalCatalog mode.",
+ "description": "The total size of all thread pools used in loading file metadata.",
"contexts": [
+ "CATALOGSERVER"
+ ],
+ "label": "Catalog Server File Metadata Loading Threads",
+ "units": "NONE",
+ "kind": "GAUGE",
+ "key": "catalog-server.metadata.file.num-loading-threads"
+ },
+ {
+ "description": "The total number of unfinished file metadata loading tasks. Each task corresponds to a partition.",
+ "contexts": [
+ "CATALOGSERVER"
+ ],
+ "label": "Catalog Server File Metadata Loading Tasks",
+ "units": "NONE",
+ "kind": "GAUGE",
+ "key": "catalog-server.metadata.file.num-loading-tasks"
+ },
+ {
+ "description": "The total number of tables that are loading file metadata.",
+ "contexts": [
+ "CATALOGSERVER"
+ ],
+ "label": "Catalog Server File Metadata Loading Tables",
+ "units": "NONE",
+ "kind": "GAUGE",
+ "key": "catalog-server.metadata.table.num-loading-file-metadata"
+ },
+ {
+ "description": "The total number of tables that are loading metadata.",
+ "contexts": [
+ "CATALOGSERVER"
+ ],
+ "label": "Catalog Server Metadata Loading Tables",
+ "units": "NONE",
+ "kind": "GAUGE",
+ "key": "catalog-server.metadata.table.num-loading-metadata"
+ },
+ {
+ "description": "The total number of tables that are loading metadata asynchronously. E.g. initial metadata loading triggered by the first access on a table.",
+ "contexts": [
+ "CATALOGSERVER"
+ ],
+ "label": "Catalog Server Async Metadata Loading Tables",
+ "units": "NONE",
+ "kind": "GAUGE",
+ "key": "catalog-server.metadata.table.async-loading.num-in-progress"
+ },
+ {
+ "description": "The total number of tables that are waiting for loading metadata asynchronously.",
+ "contexts": [
+ "CATALOGSERVER"
+ ],
+ "label": "Catalog Server Queue Size for Async Metadata Loading Tables",
+ "units": "NONE",
+ "kind": "GAUGE",
+ "key": "catalog-server.metadata.table.async-loading.queue-len"
+ },
+ {
+ "description": "The number of databases in the catalog. Untracked in LocalCatalog mode coordinators.",
+ "contexts": [
+ "CATALOGSERVER",
"IMPALAD"
],
"label": "Databases",
@@ -340,8 +401,9 @@
"key": "catalog.num-databases"
},
{
- "description": "The number of tables in the catalog. Untracked in LocalCatalog mode.",
+ "description": "The number of tables in the catalog. Untracked in LocalCatalog mode coordinators.",
"contexts": [
+ "CATALOGSERVER",
"IMPALAD"
],
"label": "Tables",
@@ -349,6 +411,36 @@
"kind": "GAUGE",
"key": "catalog.num-tables"
},
+ {
+ "description": "The number of functions in the catalog.",
+ "contexts": [
+ "CATALOGSERVER"
+ ],
+ "label": "Functions",
+ "units": "NONE",
+ "kind": "GAUGE",
+ "key": "catalog.num-functions"
+ },
+ {
+ "description": "The number of Hive Metastore Clients that are idle.",
+ "contexts": [
+ "CATALOGSERVER"
+ ],
+ "label": "Idle HMS Clients",
+ "units": "NONE",
+ "kind": "GAUGE",
+ "key": "catalog.hms-client-pool.num-idle"
+ },
+ {
+ "description": "The number of Hive Metastore Clients that are in use.",
+ "contexts": [
+ "CATALOGSERVER"
+ ],
+ "label": "HMS Clients In Use",
+ "units": "NONE",
+ "kind": "GAUGE",
+ "key": "catalog.hms-client-pool.num-in-use"
+ },
{
"description": "Catalog topic update version.",
"contexts": [
diff --git a/fe/src/main/java/org/apache/impala/catalog/Catalog.java b/fe/src/main/java/org/apache/impala/catalog/Catalog.java
index 78aae22d1..a1a1058ae 100644
--- a/fe/src/main/java/org/apache/impala/catalog/Catalog.java
+++ b/fe/src/main/java/org/apache/impala/catalog/Catalog.java
@@ -397,6 +397,14 @@ public abstract class Catalog implements AutoCloseable {
*/
public MetaStoreClient getMetaStoreClient() { return metaStoreClientPool_.getClient(); }
+ public int getNumHmsClientsIdle() {
+ return metaStoreClientPool_.getNumHmsClientsIdle();
+ }
+
+ public int getNumHmsClientsInUse() {
+ return metaStoreClientPool_.getNumHmsClientsInUse();
+ }
+
/**
* Same as the above but also update the given 'timeline'.
*/
diff --git a/fe/src/main/java/org/apache/impala/catalog/CatalogObjectCache.java b/fe/src/main/java/org/apache/impala/catalog/CatalogObjectCache.java
index d9bfc1480..dd33ab3f4 100644
--- a/fe/src/main/java/org/apache/impala/catalog/CatalogObjectCache.java
+++ b/fe/src/main/java/org/apache/impala/catalog/CatalogObjectCache.java
@@ -58,6 +58,8 @@ public class CatalogObjectCache<T extends CatalogObject> implements Iterable<T>
// all methods as synchronized.
private final Map<String, T> metadataCache_ = new ConcurrentHashMap<String, T>();
+ public int size() { return metadataCache_.size(); }
+
/**
* Adds a new catalogObject to the cache. If a catalogObject with the same name already
* exists in the cache, the new item will only be added if it has a larger catalog
diff --git a/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java b/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
index bddc3f321..9f7ae5aea 100644
--- a/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
+++ b/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
@@ -331,6 +331,12 @@ public class CatalogServiceCatalog extends Catalog {
// Table properties that require file metadata reload
private final Set<String> whitelistedTblProperties_;
+ // Total number of dbs, tables and functions in the catalog cache.
+ // Updated in each catalog topic update (getCatalogDelta()).
+ private int numDbs_ = 0;
+ private int numTables_ = 0;
+ private int numFunctions_ = 0;
+
/**
* Initialize the CatalogServiceCatalog using a given MetastoreClientPool impl.
*
@@ -663,6 +669,18 @@ public class CatalogServiceCatalog extends Catalog {
return partialObjectFetchAccess_.getQueueLength();
}
+ public int getNumAsyncWaitingTables() {
+ return tableLoadingMgr_.numRemainingItems();
+ }
+
+ public int getNumAsyncLoadingTables() {
+ return tableLoadingMgr_.numLoadsInProgress();
+ }
+
+ public int getNumDatabases() { return numDbs_; }
+ public int getNumTables() { return numTables_; }
+ public int getNumFunctions() { return numFunctions_; }
+
/**
* Adds a list of cache directive IDs for the given table name. Asynchronously
* refreshes the table metadata once all cache directives complete.
@@ -762,6 +780,11 @@ public class CatalogServiceCatalog extends Catalog {
long fromVersion;
long toVersion;
long lastResetStartVersion;
+ // Total number of dbs, tables, and functions in the catalog
+ int numDbs = 0;
+ int numTables = 0;
+ int numFunctions = 0;
+
// The keys of the updated topics.
Set<String> updatedCatalogObjects;
TSerializer serializer;
@@ -927,6 +950,7 @@ public class CatalogServiceCatalog extends Catalog {
versionLock_.readLock().unlock();
}
for (Db db: getAllDbs()) {
+ ctx.numDbs++;
addDatabaseToCatalogDelta(db, ctx);
}
for (DataSource dataSource: getAllDataSources()) {
@@ -1006,6 +1030,9 @@ public class CatalogServiceCatalog extends Catalog {
synchronized (topicUpdateLog_) {
topicUpdateLog_.notifyAll();
}
+ numDbs_ = ctx.numDbs;
+ numTables_ = ctx.numTables;
+ numFunctions_ = ctx.numFunctions;
return ctx.toVersion;
}
@@ -1272,9 +1299,11 @@ public class CatalogServiceCatalog extends Catalog {
ctx.addCatalogObject(catalogDb, false);
}
for (Table tbl: getAllTables(db)) {
+ ctx.numTables++;
addTableToCatalogDelta(tbl, ctx);
}
for (Function fn: getAllFunctions(db)) {
+ ctx.numFunctions++;
addFunctionToCatalogDelta(fn, ctx);
}
}
diff --git a/fe/src/main/java/org/apache/impala/catalog/DataSourceTable.java b/fe/src/main/java/org/apache/impala/catalog/DataSourceTable.java
index df3063948..3087a7d30 100644
--- a/fe/src/main/java/org/apache/impala/catalog/DataSourceTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/DataSourceTable.java
@@ -175,6 +175,7 @@ public class DataSourceTable extends Table implements FeDataSourceTable {
org.apache.hadoop.hive.metastore.api.Table msTbl, String reason)
throws TableLoadingException {
Preconditions.checkNotNull(msTbl);
+ Table.LOADING_TABLES.incrementAndGet();
msTable_ = msTbl;
clearColumns();
if (LOG.isTraceEnabled()) {
@@ -189,6 +190,7 @@ public class DataSourceTable extends Table implements FeDataSourceTable {
initString_ = getRequiredTableProperty(msTbl, TBL_PROP_INIT_STRING, dataSourceName);
if (msTbl.getPartitionKeysSize() > 0) {
+ Table.LOADING_TABLES.decrementAndGet();
throw new TableLoadingException("Data source table cannot contain clustering " +
"columns: " + name_);
}
@@ -205,6 +207,8 @@ public class DataSourceTable extends Table implements FeDataSourceTable {
} catch (Exception e) {
throw new TableLoadingException("Failed to load metadata for data source table: " +
name_, e);
+ } finally {
+ Table.LOADING_TABLES.decrementAndGet();
}
}
diff --git a/fe/src/main/java/org/apache/impala/catalog/Db.java b/fe/src/main/java/org/apache/impala/catalog/Db.java
index ad25324fb..6150c83a2 100644
--- a/fe/src/main/java/org/apache/impala/catalog/Db.java
+++ b/fe/src/main/java/org/apache/impala/catalog/Db.java
@@ -212,6 +212,11 @@ public class Db extends CatalogObjectImpl implements FeDb {
*/
public List<Table> getTables() { return tableCache_.getValues(); }
+ /**
+ * Returns the number of tables in this db.
+ */
+ public int getNumTables() { return tableCache_.size(); }
+
@Override
public boolean containsTable(String tableName) {
return tableCache_.contains(tableName.toLowerCase());
diff --git a/fe/src/main/java/org/apache/impala/catalog/FileMetadataLoader.java b/fe/src/main/java/org/apache/impala/catalog/FileMetadataLoader.java
index 16720e01e..a5c44245e 100644
--- a/fe/src/main/java/org/apache/impala/catalog/FileMetadataLoader.java
+++ b/fe/src/main/java/org/apache/impala/catalog/FileMetadataLoader.java
@@ -44,6 +44,7 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
import javax.annotation.Nullable;
@@ -54,6 +55,10 @@ public class FileMetadataLoader {
private final static Logger LOG = LoggerFactory.getLogger(FileMetadataLoader.class);
private static final Configuration CONF = new Configuration();
+ // The number of unfinished instances. Incremented in the constructor and decremented
+ // at the end of load().
+ public static final AtomicInteger TOTAL_TASKS = new AtomicInteger();
+
protected final Path partDir_;
protected final boolean recursive_;
protected final ImmutableMap<String, FileDescriptor> oldFdsByPath_;
@@ -104,6 +109,7 @@ public class FileMetadataLoader {
if (writeIds_ != null) {
Preconditions.checkArgument(recursive_, "ACID tables must be listed recursively");
}
+ TOTAL_TASKS.incrementAndGet();
}
public FileMetadataLoader(Path partDir, boolean recursive, List<FileDescriptor> oldFds,
@@ -161,6 +167,14 @@ public class FileMetadataLoader {
* resolved.
*/
public void load() throws CatalogException, IOException {
+ try {
+ loadInternal();
+ } finally {
+ TOTAL_TASKS.decrementAndGet();
+ }
+ }
+
+ private void loadInternal() throws CatalogException, IOException {
Preconditions.checkState(loadStats_ == null, "already loaded");
loadStats_ = new LoadStats(partDir_);
FileSystem fs = partDir_.getFileSystem(CONF);
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 2a1a5cc53..c4f293960 100644
--- a/fe/src/main/java/org/apache/impala/catalog/HBaseTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/HBaseTable.java
@@ -101,6 +101,7 @@ public class HBaseTable extends Table implements FeHBaseTable {
org.apache.hadoop.hive.metastore.api.Table msTbl, String reason)
throws TableLoadingException {
Preconditions.checkNotNull(getMetaStoreTable());
+ Table.LOADING_TABLES.incrementAndGet();
try (Timer.Context timer = getMetrics().getTimer(Table.LOAD_DURATION_METRIC).time()) {
msTable_ = msTbl;
final Timer.Context storageLoadTimer =
@@ -127,6 +128,8 @@ public class HBaseTable extends Table implements FeHBaseTable {
} catch (Exception e) {
throw new TableLoadingException("Failed to load metadata for HBase table: " + name_,
e);
+ } finally {
+ Table.LOADING_TABLES.decrementAndGet();
}
}
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 8c072844d..36324ceaa 100644
--- a/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
@@ -1252,6 +1252,7 @@ public class HdfsTable extends Table implements FeFsTable {
final Timer storageLdTimer =
getMetrics().getTimer(Table.LOAD_DURATION_STORAGE_METADATA);
storageMetadataLoadTime_ = 0;
+ Table.LOADING_TABLES.incrementAndGet();
try (ThreadNameAnnotator tna = new ThreadNameAnnotator(annotation)) {
// turn all exceptions into TableLoadingException
msTable_ = msTbl;
@@ -1317,6 +1318,7 @@ public class HdfsTable extends Table implements FeFsTable {
"warning threshold. Time: " + PrintUtils.printTimeNs(load_time_duration));
}
updateTableLoadingTime();
+ Table.LOADING_TABLES.decrementAndGet();
}
}
diff --git a/fe/src/main/java/org/apache/impala/catalog/IcebergFileMetadataLoader.java b/fe/src/main/java/org/apache/impala/catalog/IcebergFileMetadataLoader.java
index 293257b42..00e2c99d7 100644
--- a/fe/src/main/java/org/apache/impala/catalog/IcebergFileMetadataLoader.java
+++ b/fe/src/main/java/org/apache/impala/catalog/IcebergFileMetadataLoader.java
@@ -17,7 +17,9 @@
package org.apache.impala.catalog;
+import static org.apache.impala.catalog.ParallelFileMetadataLoader.TOTAL_THREADS;
import static org.apache.impala.catalog.ParallelFileMetadataLoader.createPool;
+import static org.apache.impala.catalog.ParallelFileMetadataLoader.getPoolSize;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
@@ -112,7 +114,11 @@ public class IcebergFileMetadataLoader extends FileMetadataLoader {
if (!shouldReuseOldFds()) {
super.load();
} else {
- reloadWithOldFds();
+ try {
+ reloadWithOldFds();
+ } finally {
+ FileMetadataLoader.TOTAL_TASKS.decrementAndGet();
+ }
}
}
@@ -232,19 +238,29 @@ public class IcebergFileMetadataLoader extends FileMetadataLoader {
private Map<Path, FileStatus> parallelListing(FileSystem fs) throws IOException {
String logPrefix = "Parallel Iceberg file metadata listing";
- ExecutorService pool = createPool(icebergFiles_.size(), fs, logPrefix);
+ int poolSize = getPoolSize(icebergFiles_.size(), fs);
+ ExecutorService pool = createPool(poolSize, logPrefix);
+ TOTAL_THREADS.addAndGet(poolSize);
final Set<Path> partitionPaths;
Map<Path, FileStatus> nameToFileStatus = Maps.newConcurrentMap();
try (ThreadNameAnnotator tna = new ThreadNameAnnotator(logPrefix)) {
partitionPaths = icebergFilesByPartition();
+ TOTAL_TASKS.addAndGet(partitionPaths.size());
List<Future<Void>> tasks =
partitionPaths.stream()
- .map(path -> pool.submit(() -> listingTask(fs, path, nameToFileStatus)))
+ .map(path -> pool.submit(() -> {
+ try {
+ return listingTask(fs, path, nameToFileStatus);
+ } finally {
+ TOTAL_TASKS.decrementAndGet();
+ }
+ }))
.collect(Collectors.toList());
for (Future<Void> task : tasks) { task.get(); }
} catch (ExecutionException | InterruptedException e) {
throw new IOException(String.format("%s: failed to load paths.", logPrefix), e);
} finally {
+ TOTAL_THREADS.addAndGet(-poolSize);
pool.shutdown();
}
return nameToFileStatus;
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 427f2f250..76fbb7e0f 100644
--- a/fe/src/main/java/org/apache/impala/catalog/KuduTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/KuduTable.java
@@ -331,6 +331,7 @@ public class KuduTable extends Table implements FeKuduTable {
throws TableLoadingException {
final Timer.Context context =
getMetrics().getTimer(Table.LOAD_DURATION_METRIC).time();
+ Table.LOADING_TABLES.incrementAndGet();
try {
// Copy the table to check later if anything has changed.
msTable_ = msTbl.deepCopy();
@@ -371,6 +372,7 @@ public class KuduTable extends Table implements FeKuduTable {
throw new TableLoadingException(e.getMessage());
}
} finally {
+ Table.LOADING_TABLES.decrementAndGet();
context.stop();
}
}
diff --git a/fe/src/main/java/org/apache/impala/catalog/MetaStoreClientPool.java b/fe/src/main/java/org/apache/impala/catalog/MetaStoreClientPool.java
index 8c4454ff1..fdec7fd99 100644
--- a/fe/src/main/java/org/apache/impala/catalog/MetaStoreClientPool.java
+++ b/fe/src/main/java/org/apache/impala/catalog/MetaStoreClientPool.java
@@ -19,6 +19,7 @@ package org.apache.impala.catalog;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.HiveMetaHook;
@@ -49,6 +50,9 @@ public class MetaStoreClientPool {
private static final int DEFAULT_HIVE_METASTORE_CNXN_DELAY_MS_CONF = 0;
// Maximum number of idle metastore connections in the connection pool at any point.
private static final int MAX_HMS_CONNECTION_POOL_SIZE = 32;
+
+ private final AtomicInteger numHmsClientsInUse_ = new AtomicInteger(0);
+
// Number of milliseconds to sleep between creation of HMS connections. Used to debug
// IMPALA-825.
private final int clientCreationDelayMs_;
@@ -131,6 +135,7 @@ public class MetaStoreClientPool {
public void close() {
Preconditions.checkState(isInUse_);
isInUse_ = false;
+ numHmsClientsInUse_.decrementAndGet();
// Ensure the connection isn't returned to the pool if the pool has been closed
// or if the number of connections in the pool exceeds MAX_HMS_CONNECTION_POOL_SIZE.
// This lock is needed to ensure proper behavior when a thread reads poolClosed
@@ -148,6 +153,7 @@ public class MetaStoreClientPool {
private void markInUse() {
Preconditions.checkState(!isInUse_);
isInUse_ = true;
+ numHmsClientsInUse_.incrementAndGet();
}
}
@@ -225,4 +231,7 @@ public class MetaStoreClientPool {
client.getHiveClient().close();
}
}
+
+ public int getNumHmsClientsIdle() { return clientPool_.size(); }
+ public int getNumHmsClientsInUse() { return numHmsClientsInUse_.get(); }
}
diff --git a/fe/src/main/java/org/apache/impala/catalog/ParallelFileMetadataLoader.java b/fe/src/main/java/org/apache/impala/catalog/ParallelFileMetadataLoader.java
index f0fb5871a..4ee6beaed 100644
--- a/fe/src/main/java/org/apache/impala/catalog/ParallelFileMetadataLoader.java
+++ b/fe/src/main/java/org/apache/impala/catalog/ParallelFileMetadataLoader.java
@@ -26,6 +26,7 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicInteger;
import javax.annotation.Nullable;
import org.apache.hadoop.fs.FileSystem;
@@ -65,6 +66,9 @@ public class ParallelFileMetadataLoader {
private static final int MAX_NON_HDFS_PARTITIONS_PARALLEL_LOAD =
BackendConfig.INSTANCE.maxNonHdfsPartsParallelLoad();
+ public static final AtomicInteger TOTAL_THREADS = new AtomicInteger(0);
+ public static final AtomicInteger TOTAL_TABLES = new AtomicInteger(0);
+
// Maximum number of errors logged when loading partitioned tables.
private static final int MAX_PATH_METADATA_LOADING_ERRORS_TO_LOG = 100;
@@ -169,8 +173,11 @@ public class ParallelFileMetadataLoader {
if (loaders_.isEmpty()) return;
int failedLoadTasks = 0;
- ExecutorService pool = createPool(loaders_.size(), fs_, logPrefix_);
+ int poolSize = getPoolSize(loaders_.size(), fs_);
+ ExecutorService pool = createPool(poolSize, logPrefix_);
+ TOTAL_THREADS.addAndGet(poolSize);
try (ThreadNameAnnotator tna = new ThreadNameAnnotator(logPrefix_)) {
+ TOTAL_TABLES.incrementAndGet();
List<Pair<FileMetadataLoader, Future<Void>>> futures =
new ArrayList<>(loaders_.size());
for (FileMetadataLoader loader : loaders_.values()) {
@@ -191,6 +198,8 @@ public class ParallelFileMetadataLoader {
}
} finally {
pool.shutdown();
+ TOTAL_THREADS.addAndGet(-poolSize);
+ TOTAL_TABLES.addAndGet(-1);
}
if (failedLoadTasks > 0) {
int errorsNotLogged = failedLoadTasks - MAX_PATH_METADATA_LOADING_ERRORS_TO_LOG;
@@ -204,7 +213,21 @@ public class ParallelFileMetadataLoader {
}
/**
- * Returns the thread pool to load the file metadata.
+ * Returns the thread pool to load the file metadata. Callers should use
+ * {@link #getPoolSize(int, FileSystem)} to get a correct pool size.
+ */
+ public static ExecutorService createPool(int poolSize, String logPrefix) {
+ Preconditions.checkState(poolSize > 0, "Illegal poolSize: {}", poolSize);
+ if (poolSize == 1) {
+ return MoreExecutors.newDirectExecutorService();
+ } else {
+ LOG.info("{} using a thread pool of size {}", logPrefix, poolSize);
+ return Executors.newFixedThreadPool(poolSize);
+ }
+ }
+
+ /**
+ * Returns the thread pool size to load the file metadata.
*
* We use different thread pool sizes for HDFS and non-HDFS tables since the latter
* supports much higher throughput of RPC calls for listStatus/listFiles. For
@@ -215,19 +238,12 @@ public class ParallelFileMetadataLoader {
* clusters. We narrowed it down to scalability bottlenecks in HDFS RPC implementation
* (HADOOP-14558) on both the server and the client side.
*/
- public static ExecutorService createPool(
- int numLoaders, FileSystem fs, String logPrefix) {
- Preconditions.checkState(numLoaders > 0);
+ public static int getPoolSize(int numLoaders, FileSystem fs) {
int poolSize = FileSystemUtil.supportsStorageIds(fs) ?
MAX_HDFS_PARTITIONS_PARALLEL_LOAD :
MAX_NON_HDFS_PARTITIONS_PARALLEL_LOAD;
// Thread pool size need not exceed the number of paths to be loaded.
poolSize = Math.min(numLoaders, poolSize);
- if (poolSize == 1) {
- return MoreExecutors.newDirectExecutorService();
- } else {
- LOG.info("{} using a thread pool of size {}", logPrefix, poolSize);
- return Executors.newFixedThreadPool(poolSize);
- }
+ return poolSize;
}
}
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 3f95764c1..5af181169 100644
--- a/fe/src/main/java/org/apache/impala/catalog/Table.java
+++ b/fe/src/main/java/org/apache/impala/catalog/Table.java
@@ -24,6 +24,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock;
@@ -193,6 +194,8 @@ public abstract class Table extends CatalogObjectImpl implements FeTable {
public static final String TBL_PROP_EXTERNAL_TABLE_PURGE = "external.table.purge";
public static final String TBL_PROP_EXTERNAL_TABLE_PURGE_DEFAULT = "TRUE";
+ public static final AtomicInteger LOADING_TABLES = new AtomicInteger(0);
+
// this field represents the last event id in metastore upto which this table is
// synced. It is used if the flag sync_to_latest_event_on_ddls is set to true.
// Making it as volatile so that read and write of this variable are thread safe.
diff --git a/fe/src/main/java/org/apache/impala/catalog/TableLoadingMgr.java b/fe/src/main/java/org/apache/impala/catalog/TableLoadingMgr.java
index 48720d0bb..cb9ec2df5 100644
--- a/fe/src/main/java/org/apache/impala/catalog/TableLoadingMgr.java
+++ b/fe/src/main/java/org/apache/impala/catalog/TableLoadingMgr.java
@@ -365,4 +365,12 @@ public class TableLoadingMgr {
}
}
}
+
+ public int numRemainingItems() {
+ return tableLoadingDeque_.size();
+ }
+
+ public int numLoadsInProgress() {
+ return loadingTables_.size();
+ }
}
diff --git a/fe/src/main/java/org/apache/impala/catalog/View.java b/fe/src/main/java/org/apache/impala/catalog/View.java
index d4a3324c8..c12f34c63 100644
--- a/fe/src/main/java/org/apache/impala/catalog/View.java
+++ b/fe/src/main/java/org/apache/impala/catalog/View.java
@@ -86,6 +86,7 @@ public class View extends Table implements FeView {
org.apache.hadoop.hive.metastore.api.Table msTbl, String reason)
throws TableLoadingException {
try {
+ Table.LOADING_TABLES.incrementAndGet();
clearColumns();
msTable_ = msTbl;
// Load columns.
@@ -106,6 +107,8 @@ public class View extends Table implements FeView {
throw e;
} catch (Exception e) {
throw new TableLoadingException("Failed to load metadata for view: " + name_, e);
+ } finally {
+ Table.LOADING_TABLES.decrementAndGet();
}
}
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 afe40bfd1..fe300fa48 100644
--- a/fe/src/main/java/org/apache/impala/service/JniCatalog.java
+++ b/fe/src/main/java/org/apache/impala/service/JniCatalog.java
@@ -36,9 +36,12 @@ import org.apache.impala.catalog.CatalogException;
import org.apache.impala.catalog.CatalogServiceCatalog;
import org.apache.impala.catalog.Db;
import org.apache.impala.catalog.FeDb;
+import org.apache.impala.catalog.FileMetadataLoader;
import org.apache.impala.catalog.Function;
import org.apache.impala.catalog.MetaStoreClientPool;
import org.apache.impala.catalog.MetaStoreClientPool.MetaStoreClient;
+import org.apache.impala.catalog.ParallelFileMetadataLoader;
+import org.apache.impala.catalog.Table;
import org.apache.impala.catalog.events.ExternalEventsProcessor;
import org.apache.impala.catalog.events.MetastoreEvents.EventFactoryForSyncToLatestEvent;
import org.apache.impala.catalog.events.MetastoreEvents.MetastoreEventFactory;
@@ -532,6 +535,22 @@ public class JniCatalog {
"getCatalogServerMetrics", shortDesc, () -> {
response.setCatalog_partial_fetch_rpc_queue_len(
catalog_.getPartialFetchRpcQueueLength());
+ response.setCatalog_num_file_metadata_loading_threads(
+ ParallelFileMetadataLoader.TOTAL_THREADS.get());
+ response.setCatalog_num_tables_loading_file_metadata(
+ ParallelFileMetadataLoader.TOTAL_TABLES.get());
+ response.setCatalog_num_file_metadata_loading_tasks(
+ FileMetadataLoader.TOTAL_TASKS.get());
+ response.setCatalog_num_tables_loading_metadata(Table.LOADING_TABLES.get());
+ response.setCatalog_num_tables_async_loading_metadata(
+ catalog_.getNumAsyncLoadingTables());
+ response.setCatalog_num_tables_waiting_for_async_loading(
+ catalog_.getNumAsyncWaitingTables());
+ response.setCatalog_num_dbs(catalog_.getNumDatabases());
+ response.setCatalog_num_tables(catalog_.getNumTables());
+ response.setCatalog_num_functions(catalog_.getNumFunctions());
+ response.setCatalog_num_hms_clients_idle(catalog_.getNumHmsClientsIdle());
+ response.setCatalog_num_hms_clients_in_use(catalog_.getNumHmsClientsInUse());
response.setEvent_metrics(catalog_.getEventProcessorMetrics());
return response;
});
diff --git a/tests/webserver/test_web_pages.py b/tests/webserver/test_web_pages.py
index 74db29fb1..553ce0218 100644
--- a/tests/webserver/test_web_pages.py
+++ b/tests/webserver/test_web_pages.py
@@ -898,6 +898,23 @@ class TestWebPage(ImpalaTestSuite):
page = requests.head("http://localhost:25020/operations")
assert page.status_code == requests.codes.ok
+ def test_catalog_metrics(self):
+ """Test /metrics of catalogd"""
+ url = self.METRICS_URL.format(*self.CATALOG_TEST_PORT) + "?json"
+ json_res = json.loads(requests.get(url).text)
+ metric_keys = {m["name"] for m in json_res["metric_group"]["metrics"]}
+ assert "catalog-server.metadata.file.num-loading-threads" in metric_keys
+ assert "catalog-server.metadata.file.num-loading-tasks" in metric_keys
+ assert "catalog-server.metadata.table.num-loading-file-metadata" in metric_keys
+ assert "catalog-server.metadata.table.num-loading-metadata" in metric_keys
+ assert "catalog-server.metadata.table.async-loading.num-in-progress" in metric_keys
+ assert "catalog-server.metadata.table.async-loading.queue-len" in metric_keys
+ assert "catalog.num-databases" in metric_keys
+ assert "catalog.num-tables" in metric_keys
+ assert "catalog.num-functions" in metric_keys
+ assert "catalog.hms-client-pool.num-idle" in metric_keys
+ assert "catalog.hms-client-pool.num-in-use" in metric_keys
+
def test_query_progress(self):
"""Tests that /queries page shows query progress."""
query = "select count(*) from functional_parquet.alltypes where bool_col = sleep(100)"