You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by bh...@apache.org on 2018/10/19 00:56:45 UTC
impala git commit: IMPALA-7597: wraps retries around Frontend
metadata operations.
Repository: impala
Updated Branches:
refs/heads/master 9f5c5e6df -> 5cc49c343
IMPALA-7597: wraps retries around Frontend metadata operations.
When configured to use the local catalog, concurrent metadata
reads and writes can cause the CatalogMetaProvider to throw
an InconsistentMetadataFetchException. Queries have been wrapped
with a retry loop, but the other frontend methods, such listing
table or partition information, can also fail from the same error.
These errors were seen under a workload consisting of concurrent
adding and showing partitions.
This change wraps call-sites (primarily in Frontend.java) that acquire
a Catalog, so have a chance of throwing an InconsistentMetadataFetchExecption.
Testing:
- added a test that checks whether inconsistent metadata exceptions
are seen in a concurrent workload.
Change-Id: I43a21571d54a7966c5c68bea1fe69dbc62be2a0b
Reviewed-on: http://gerrit.cloudera.org:8080/11608
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/impala/commit/5cc49c34
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/5cc49c34
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/5cc49c34
Branch: refs/heads/master
Commit: 5cc49c343f8558602af2663e3bb519da6d9852cc
Parents: 9f5c5e6
Author: Vuk Ercegovac <ve...@cloudera.com>
Authored: Mon Oct 1 15:43:02 2018 -0700
Committer: Impala Public Jenkins <im...@cloudera.com>
Committed: Fri Oct 19 00:20:12 2018 +0000
----------------------------------------------------------------------
.../org/apache/impala/service/Frontend.java | 135 ++++++++++-
.../org/apache/impala/service/MetadataOp.java | 18 ++
tests/custom_cluster/test_local_catalog.py | 237 +++++++++++--------
3 files changed, 289 insertions(+), 101 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/5cc49c34/fe/src/main/java/org/apache/impala/service/Frontend.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/service/Frontend.java b/fe/src/main/java/org/apache/impala/service/Frontend.java
index d31b573..8aeafb1 100644
--- a/fe/src/main/java/org/apache/impala/service/Frontend.java
+++ b/fe/src/main/java/org/apache/impala/service/Frontend.java
@@ -586,6 +586,20 @@ public class Frontend {
*/
public TLoadDataResp loadTableData(TLoadDataReq request) throws ImpalaException,
IOException {
+ TTableName tableName = request.getTable_name();
+ RetryTracker retries = new RetryTracker(
+ String.format("load table data %s.%s", tableName.db_name, tableName.table_name));
+ while (true) {
+ try {
+ return doLoadTableData(request);
+ } catch(InconsistentMetadataFetchException e) {
+ retries.handleRetryOrThrow(e);
+ }
+ }
+ }
+
+ private TLoadDataResp doLoadTableData(TLoadDataReq request) throws ImpalaException,
+ IOException {
TableName tableName = TableName.fromThrift(request.getTable_name());
// Get the destination for the load. If the load is targeting a partition,
@@ -657,11 +671,60 @@ public class Frontend {
}
/**
+ * Keeps track of retries when handling InconsistentMetadataFetchExceptions.
+ * Whenever a Catalog object is acquired (e.g., getCatalog), operations that access
+ * finer-grained objects, such as tables and partitions, can throw such a runtime
+ * exception. Inconsistent metadata comes up due to interleaving catalog object updates
+ * with retrieving those objects. Instead of bubbling up the issue to the user, retrying
+ * can get the user's operation to run on a consistent snapshot and to succeed.
+ * Retries are *not* needed for accessing top-level objects such as databases, since
+ * they do not have a parent, so cannot be inconsistent.
+ * TODO: this class is typically used in a loop at the call-site. replace with lambdas
+ * in Java 8 to simplify the looping boilerplate.
+ */
+ public static class RetryTracker {
+ // Number of exceptions seen
+ private int attempt_ = 0;
+ // Message to add when logging retries.
+ private final String msg_;
+
+ public RetryTracker(String msg) { msg_ = msg; }
+
+ /**
+ * Record a retry. If the number of retries exceeds to configured maximum, the
+ * exception is thrown. Otherwise, the number of retries is incremented and logged.
+ * TODO: record these retries in the profile as done for query retries.
+ */
+ public void handleRetryOrThrow(InconsistentMetadataFetchException exception) {
+ if (attempt_++ >= INCONSISTENT_METADATA_NUM_RETRIES) throw exception;
+ if (attempt_ > 1) {
+ // Back-off a bit on later retries.
+ Uninterruptibles.sleepUninterruptibly(200 * attempt_, TimeUnit.MILLISECONDS);
+ }
+ LOG.warn(String.format("Retried %s: (retry #%s of %s)", msg_,
+ attempt_, INCONSISTENT_METADATA_NUM_RETRIES), exception);
+ }
+ }
+
+ /**
* Returns all tables in database 'dbName' that match the pattern of 'matcher' and are
* accessible to 'user'.
*/
public List<String> getTableNames(String dbName, PatternMatcher matcher,
User user) throws ImpalaException {
+ RetryTracker retries = new RetryTracker(
+ String.format("fetching %s table names", dbName));
+ while (true) {
+ try {
+ return doGetTableNames(dbName, matcher, user);
+ } catch(InconsistentMetadataFetchException e) {
+ retries.handleRetryOrThrow(e);
+ }
+ }
+ }
+
+ private List<String> doGetTableNames(String dbName, PatternMatcher matcher,
+ User user) throws ImpalaException {
List<String> tblNames = getCatalog().getTableNames(dbName, matcher);
if (authzConfig_.isEnabled()) {
Iterator<String> iter = tblNames.iterator();
@@ -738,6 +801,7 @@ public class Frontend {
* matches all data sources.
*/
public List<? extends FeDataSource> getDataSrcs(String pattern) {
+ // TODO: handle InconsistentMetadataException for data sources.
return getCatalog().getDataSources(
PatternMatcher.createHivePatternMatcher(pattern));
}
@@ -747,6 +811,19 @@ public class Frontend {
*/
public TResultSet getColumnStats(String dbName, String tableName)
throws ImpalaException {
+ RetryTracker retries = new RetryTracker(
+ String.format("fetching column stats from %s.%s", dbName, tableName));
+ while (true) {
+ try {
+ return doGetColumnStats(dbName, tableName);
+ } catch(InconsistentMetadataFetchException e) {
+ retries.handleRetryOrThrow(e);
+ }
+ }
+ }
+
+ private TResultSet doGetColumnStats(String dbName, String tableName)
+ throws ImpalaException {
FeTable table = getCatalog().getTable(dbName, tableName);
TResultSet result = new TResultSet();
TResultSetMetadata resultSchema = new TResultSetMetadata();
@@ -775,6 +852,19 @@ public class Frontend {
*/
public TResultSet getTableStats(String dbName, String tableName, TShowStatsOp op)
throws ImpalaException {
+ RetryTracker retries = new RetryTracker(
+ String.format("fetching table stats from %s.%s", dbName, tableName));
+ while (true) {
+ try {
+ return doGetTableStats(dbName, tableName, op);
+ } catch(InconsistentMetadataFetchException e) {
+ retries.handleRetryOrThrow(e);
+ }
+ }
+ }
+
+ private TResultSet doGetTableStats(String dbName, String tableName, TShowStatsOp op)
+ throws ImpalaException {
FeTable table = getCatalog().getTable(dbName, tableName);
if (table instanceof FeFsTable) {
return ((FeFsTable) table).getTableStats();
@@ -801,6 +891,20 @@ public class Frontend {
public List<Function> getFunctions(TFunctionCategory category,
String dbName, String fnPattern, boolean exactMatch)
throws DatabaseNotFoundException {
+ RetryTracker retries = new RetryTracker(
+ String.format("fetching functions from %s", dbName));
+ while (true) {
+ try {
+ return doGetFunctions(category, dbName, fnPattern, exactMatch);
+ } catch(InconsistentMetadataFetchException e) {
+ retries.handleRetryOrThrow(e);
+ }
+ }
+ }
+
+ private List<Function> doGetFunctions(TFunctionCategory category,
+ String dbName, String fnPattern, boolean exactMatch)
+ throws DatabaseNotFoundException {
FeDb db = getCatalog().getDb(dbName);
if (db == null) {
throw new DatabaseNotFoundException("Database '" + dbName + "' not found");
@@ -840,7 +944,21 @@ public class Frontend {
*/
public TDescribeResult describeTable(TTableName tableName,
TDescribeOutputStyle outputStyle, User user) throws ImpalaException {
- FeTable table = getCatalog().getTable(tableName.db_name, tableName.table_name);
+ RetryTracker retries = new RetryTracker(
+ String.format("fetching table %s.%s", tableName.db_name, tableName.table_name));
+ while (true) {
+ try {
+ return doDescribeTable(tableName, outputStyle, user);
+ } catch(InconsistentMetadataFetchException e) {
+ retries.handleRetryOrThrow(e);
+ }
+ }
+ }
+
+ private TDescribeResult doDescribeTable(TTableName tableName,
+ TDescribeOutputStyle outputStyle, User user) throws ImpalaException {
+ FeTable table = getCatalog().getTable(tableName.db_name,
+ tableName.table_name);
List<Column> filteredColumns;
if (authzConfig_.isEnabled()) {
// First run a table check
@@ -1360,6 +1478,21 @@ public class Frontend {
* Returns all files info of a table or partition.
*/
public TResultSet getTableFiles(TShowFilesParams request)
+ throws ImpalaException {
+ TTableName tableName = request.getTable_name();
+ RetryTracker retries = new RetryTracker(
+ String.format("getting table files %s.%s", tableName.db_name,
+ tableName.table_name));
+ while (true) {
+ try {
+ return doGetTableFiles(request);
+ } catch(InconsistentMetadataFetchException e) {
+ retries.handleRetryOrThrow(e);
+ }
+ }
+ }
+
+ private TResultSet doGetTableFiles(TShowFilesParams request)
throws ImpalaException{
FeTable table = getCatalog().getTable(request.getTable_name().getDb_name(),
request.getTable_name().getTable_name());
http://git-wip-us.apache.org/repos/asf/impala/blob/5cc49c34/fe/src/main/java/org/apache/impala/service/MetadataOp.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/service/MetadataOp.java b/fe/src/main/java/org/apache/impala/service/MetadataOp.java
index d221e2c..63484b1 100644
--- a/fe/src/main/java/org/apache/impala/service/MetadataOp.java
+++ b/fe/src/main/java/org/apache/impala/service/MetadataOp.java
@@ -35,6 +35,7 @@ import org.apache.impala.catalog.Function;
import org.apache.impala.catalog.PrimitiveType;
import org.apache.impala.catalog.ScalarType;
import org.apache.impala.catalog.Type;
+import org.apache.impala.catalog.local.InconsistentMetadataFetchException;
import org.apache.impala.common.ImpalaException;
import org.apache.impala.thrift.TColumn;
import org.apache.impala.thrift.TColumnValue;
@@ -258,6 +259,23 @@ public class MetadataOp {
PatternMatcher schemaPatternMatcher, PatternMatcher tablePatternMatcher,
PatternMatcher columnPatternMatcher, PatternMatcher fnPatternMatcher, User user)
throws ImpalaException {
+ Frontend.RetryTracker retries = new Frontend.RetryTracker(
+ String.format("fetching metadata"));
+ while (true) {
+ try {
+ return doGetDbsMetadata(fe, catalogName,
+ schemaPatternMatcher, tablePatternMatcher,
+ columnPatternMatcher, fnPatternMatcher, user);
+ } catch(InconsistentMetadataFetchException e) {
+ retries.handleRetryOrThrow(e);
+ }
+ }
+ }
+
+ private static DbsMetadata doGetDbsMetadata(Frontend fe, String catalogName,
+ PatternMatcher schemaPatternMatcher, PatternMatcher tablePatternMatcher,
+ PatternMatcher columnPatternMatcher, PatternMatcher fnPatternMatcher, User user)
+ throws ImpalaException {
DbsMetadata result = new DbsMetadata();
// Hive does not have a catalog concept. Returns nothing if the request specifies an
http://git-wip-us.apache.org/repos/asf/impala/blob/5cc49c34/tests/custom_cluster/test_local_catalog.py
----------------------------------------------------------------------
diff --git a/tests/custom_cluster/test_local_catalog.py b/tests/custom_cluster/test_local_catalog.py
index 916443b..3815f34 100644
--- a/tests/custom_cluster/test_local_catalog.py
+++ b/tests/custom_cluster/test_local_catalog.py
@@ -44,7 +44,6 @@ class TestCompactCatalogUpdates(CustomClusterTestSuite):
return dict(metrics_data)
assert False, "Catalog cache metrics not found in %s" % child_groups
-
@pytest.mark.execute_serially
@CustomClusterTestSuite.with_args(
impalad_args="--use_local_catalog=true",
@@ -172,6 +171,142 @@ class TestCompactCatalogUpdates(CustomClusterTestSuite):
@CustomClusterTestSuite.with_args(
impalad_args="--use_local_catalog=true",
catalogd_args="--catalog_topic_mode=minimal")
+ def test_cache_metrics(self, unique_database):
+ """
+ Test that profile output includes impalad local cache metrics. Also verifies that
+ the daemon level metrics are updated between query runs.
+ """
+ try:
+ impalad = self.cluster.impalads[0]
+ client = impalad.service.create_beeswax_client()
+ cache_hit_rate_metric_key = "catalog.cache.hit-rate"
+ cache_miss_rate_metric_key = "catalog.cache.miss-rate"
+ cache_hit_count_metric_key = "catalog.cache.hit-count"
+ cache_request_count_metric_key = "catalog.cache.request-count"
+ cache_request_count_prev_run = 0
+ cache_hit_count_prev_run = 0
+ test_table_name = "%s.test_cache_metrics_test_tbl" % unique_database
+ # A mix of queries of various types.
+ queries_to_test = ["select count(*) from functional.alltypes",
+ "explain select count(*) from functional.alltypes",
+ "create table %s (a int)" % test_table_name,
+ "drop table %s" % test_table_name]
+ for _ in xrange(0, 10):
+ for query in queries_to_test:
+ ret = self.execute_query_expect_success(client, query)
+ assert ret.runtime_profile.count("Frontend:") == 1
+ assert ret.runtime_profile.count("CatalogFetch") > 1
+ cache_metrics = self.get_catalog_cache_metrics(impalad)
+ cache_hit_rate = cache_metrics[cache_hit_rate_metric_key]
+ cache_miss_rate = cache_metrics[cache_miss_rate_metric_key]
+ cache_hit_count = cache_metrics[cache_hit_count_metric_key]
+ cache_request_count = cache_metrics[cache_request_count_metric_key]
+ assert cache_hit_rate > 0.0 and cache_hit_rate < 1.0
+ assert cache_miss_rate > 0.0 and cache_miss_rate < 1.0
+ assert cache_hit_count > cache_hit_count_prev_run,\
+ "%s not updated between two query runs, query - %s"\
+ % (cache_hit_count_metric_key, query)
+ assert cache_request_count > cache_request_count_prev_run,\
+ "%s not updated betweeen two query runs, query - %s"\
+ % (cache_request_count_metric_key, query)
+ cache_hit_count_prev_run = cache_hit_count
+ cache_request_count_prev_run = cache_request_count
+ finally:
+ client.close()
+
+class TestLocalCatalogRetries(CustomClusterTestSuite):
+
+ def _check_metadata_retries(self, queries):
+ """
+ Runs 'queries' concurrently, recording any inconsistent metadata exceptions.
+ 'queries' is a list of query strings. The queries are run by two threads,
+ each one selecting a random query to run in a loop.
+ """
+ # Tracks number of inconsistent metadata exceptions.
+ inconsistent_seen = [0]
+ inconsistent_seen_lock = threading.Lock()
+ # Tracks query failures for all other reasons.
+ failed_queries = Queue.Queue()
+ try:
+ client1 = self.cluster.impalads[0].service.create_beeswax_client()
+ client2 = self.cluster.impalads[1].service.create_beeswax_client()
+
+ def stress_thread(client):
+ # Loops, picks a random query in each iteration, runs it,
+ # and looks for retries and InconsistentMetadataFetchExceptions.
+ # Limit the number of queries run. Too many queries takes up too
+ # much memory which can get OOM killed in tests.
+ attempt = 0
+ while inconsistent_seen[0] == 0 and attempt < 100:
+ q = random.choice(queries)
+ attempt += 1
+ try:
+ ret = self.execute_query_unchecked(client, q)
+ except Exception, e:
+ if 'InconsistentMetadataFetchException' in str(e):
+ with inconsistent_seen_lock:
+ inconsistent_seen[0] += 1
+ else:
+ failed_queries.put((q, str(e)))
+
+ threads = [threading.Thread(target=stress_thread, args=(c,))
+ for c in [client1, client2]]
+ for t in threads:
+ t.start()
+ for t in threads:
+ # When there are failures, they're observed quickly.
+ t.join(30)
+
+ assert failed_queries.empty(),\
+ "Failed query count non zero: %s" % list(failed_queries.queue)
+
+ finally:
+ client1.close()
+ client2.close()
+ return inconsistent_seen[0]
+
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args(
+ impalad_args="--use_local_catalog=true",
+ catalogd_args="--catalog_topic_mode=minimal")
+ def test_fetch_metadata_retry(self):
+ """
+ Tests that operations that fetch metadata (excluding those fetches needed for
+ query planning) retry when they hit an InconsistentMetadataFetchException.
+ """
+ queries = [
+ "show column stats functional.alltypes",
+ "show table stats functional.alltypes",
+ "describe extended functional.alltypes",
+ "show tables in functional like 'all*'",
+ "show files in functional.alltypes",
+ "refresh functional.alltypes"]
+ seen = self._check_metadata_retries(queries)
+ assert seen == 0, "Saw inconsistent metadata"
+
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args(
+ impalad_args="--use_local_catalog=true --local_catalog_max_fetch_retries=0",
+ catalogd_args="--catalog_topic_mode=minimal")
+ def test_replan_limit(self):
+ """
+ Tests that the flag to limit the number of retries works and that
+ an inconsistent metadata exception when running concurrent reads/writes
+ is seen. With the max retries set to 0, no retries are expected and with
+ the concurrent read/write workload, an inconsistent metadata exception is
+ expected.
+ """
+ queries = [
+ 'refresh functional.alltypes',
+ 'select count(*) from functional.alltypes where month=4',
+ 'select count(*) from functional.alltypes where month=5']
+ seen = self._check_metadata_retries(queries)
+ assert seen > 0, "Did not observe inconsistent metadata"
+
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args(
+ impalad_args="--use_local_catalog=true",
+ catalogd_args="--catalog_topic_mode=minimal")
def test_replan_on_stale_metadata(self, unique_database):
"""
Tests that when metadata is inconsistent while planning a query,
@@ -201,6 +336,7 @@ class TestCompactCatalogUpdates(CustomClusterTestSuite):
# indicates that a replan has happened.
# We expect stress_thread to cause a re-plan. The counter is stored in a
# mutable container so that stress_thread can update it.
+ # TODO: consolidate with _check_metadata_retries.
replans_seen = [0]
replans_seen_lock = threading.Lock()
@@ -278,102 +414,3 @@ class TestCompactCatalogUpdates(CustomClusterTestSuite):
finally:
client1.close()
client2.close()
-
-
- @pytest.mark.execute_serially
- @CustomClusterTestSuite.with_args(
- impalad_args="--use_local_catalog=true --local_catalog_max_fetch_retries=0",
- catalogd_args="--catalog_topic_mode=minimal")
- def test_replan_limit(self, unique_database):
- """
- Tests that the flag to limit the number of retries works and that
- an inconsistent metadata exception when running concurrent reads/writes
- is seen. With the max retries set to 0, no retries are expected and with
- the concurrent read/write workload, an inconsistent metadata exception is
- expected.
- """
- try:
- client1 = self.cluster.impalads[0].service.create_beeswax_client()
- client2 = self.cluster.impalads[1].service.create_beeswax_client()
-
- # Keeps track of the inconsistent exceptions seen. A mutable container
- # is used by multiple threads to update this state.
- inconsistency_seen = [0]
- inconsistency_seen_lock = threading.Lock()
-
- def stress_thread(client):
- # Each thread picks one of these queries. The queries, accessing
- # a subset of partitions, will detect version skew when the refresh
- # is concurrently run. An inconsistent metadata exception will be
- # thrown because of the version skew.
- while inconsistency_seen[0] == 0:
- q = random.choice([
- 'refresh functional.alltypes',
- 'select count(*) from functional.alltypes where month=4',
- 'select count(*) from functional.alltypes where month=5'])
- try:
- ret = self.execute_query_unchecked(client, q)
- # Since the max retries is 0, we should never retry.
- assert RETRY_PROFILE_MSG not in ret.runtime_profile
- except Exception as e:
- if 'InconsistentMetadataFetchException' in str(e):
- with inconsistency_seen_lock:
- inconsistency_seen[0] += 1
-
- threads = [threading.Thread(target=stress_thread, args=(c,))
- for c in [client1, client2]]
- for t in threads:
- t.start()
- for t in threads:
- t.join(10)
- assert inconsistency_seen[0] > 0, "Did not observe inconsistent metadata"
- finally:
- client1.close()
- client2.close()
-
- @pytest.mark.execute_serially
- @CustomClusterTestSuite.with_args(
- impalad_args="--use_local_catalog=true",
- catalogd_args="--catalog_topic_mode=minimal")
- def test_cache_metrics(self, unique_database):
- """
- Test that profile output includes impalad local cache metrics. Also verifies that
- the daemon level metrics are updated between query runs.
- """
- try:
- impalad = self.cluster.impalads[0]
- client = impalad.service.create_beeswax_client()
- cache_hit_rate_metric_key = "catalog.cache.hit-rate"
- cache_miss_rate_metric_key = "catalog.cache.miss-rate"
- cache_hit_count_metric_key = "catalog.cache.hit-count"
- cache_request_count_metric_key = "catalog.cache.request-count"
- cache_request_count_prev_run = 0
- cache_hit_count_prev_run = 0
- test_table_name = "%s.test_cache_metrics_test_tbl" % unique_database
- # A mix of queries of various types.
- queries_to_test = ["select count(*) from functional.alltypes",
- "explain select count(*) from functional.alltypes",
- "create table %s (a int)" % test_table_name,
- "drop table %s" % test_table_name]
- for _ in xrange(0, 10):
- for query in queries_to_test:
- ret = self.execute_query_expect_success(client, query)
- assert ret.runtime_profile.count("Frontend:") == 1
- assert ret.runtime_profile.count("CatalogFetch") > 1
- cache_metrics = self.get_catalog_cache_metrics(impalad)
- cache_hit_rate = cache_metrics[cache_hit_rate_metric_key]
- cache_miss_rate = cache_metrics[cache_miss_rate_metric_key]
- cache_hit_count = cache_metrics[cache_hit_count_metric_key]
- cache_request_count = cache_metrics[cache_request_count_metric_key]
- assert cache_hit_rate > 0.0 and cache_hit_rate < 1.0
- assert cache_miss_rate > 0.0 and cache_miss_rate < 1.0
- assert cache_hit_count > cache_hit_count_prev_run,\
- "%s not updated between two query runs, query - %s"\
- % (cache_hit_count_metric_key, query)
- assert cache_request_count > cache_request_count_prev_run,\
- "%s not updated betweeen two query runs, query - %s"\
- % (cache_request_count_metric_key, query)
- cache_hit_count_prev_run = cache_hit_count
- cache_request_count_prev_run = cache_request_count
- finally:
- client.close()