You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by tm...@apache.org on 2018/09/18 00:37:13 UTC
[2/2] impala git commit: IMPALA-7530,
IMPALA-7509. Re-plan queries if they fetch inconsistent metadata
IMPALA-7530, IMPALA-7509. Re-plan queries if they fetch inconsistent metadata
Given the granular caching in CatalogdMetaProvider, it's possible to
have a situation like the following:
- impalad has cached the table list including some table "t"
- some other daemon has issued deletion of "t"
- we try to access "t" before the invalidation has reached us via the
statestore
In this case, we'll get an error back when we try to fetch the table
"t". This can confuse the planning process since it previously
determined that the table exists. Note that this may occur either when
"t" is first referenced or later during planning (eg when fetching a
specific partition post-pruning), so it wouldn't be possible to simply
convert it into a 'table does not exist' result.
The solution here is to throw InconsistentMetadataFetchException after
invalidating the table list and associated object. This then triggers a
re-plan of the query.
This patch implements such re-planning up to 10 times before eventually
giving up. Given that each attempt invalidates the inconsistent cache we
would normally expect it to succeed on the first such retry. The limit
of 10 retries is just to avoid infinite loops in the case of a bug.
Change-Id: I17389954780fa22d7866224c17e128458fffa545
Reviewed-on: http://gerrit.cloudera.org:8080/11403
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/94cfdac6
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/94cfdac6
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/94cfdac6
Branch: refs/heads/master
Commit: 94cfdac664cbc9644c4873407783727ff0b87710
Parents: 7eb64a9
Author: Todd Lipcon <to...@apache.org>
Authored: Fri Sep 7 16:11:04 2018 -0700
Committer: Impala Public Jenkins <im...@cloudera.com>
Committed: Mon Sep 17 22:00:04 2018 +0000
----------------------------------------------------------------------
common/thrift/CatalogService.thrift | 20 +++-
.../impala/catalog/CatalogServiceCatalog.java | 34 ++++--
.../catalog/local/CatalogdMetaProvider.java | 107 ++++++++++---------
.../org/apache/impala/service/Frontend.java | 33 +++++-
tests/custom_cluster/test_local_catalog.py | 66 +++++++++++-
5 files changed, 192 insertions(+), 68 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/94cfdac6/common/thrift/CatalogService.thrift
----------------------------------------------------------------------
diff --git a/common/thrift/CatalogService.thrift b/common/thrift/CatalogService.thrift
index 2a4edb7..6b94697 100644
--- a/common/thrift/CatalogService.thrift
+++ b/common/thrift/CatalogService.thrift
@@ -374,19 +374,29 @@ struct TGetPartialCatalogObjectRequest {
5: optional TCatalogInfoSelector catalog_info_selector
}
+enum CatalogLookupStatus {
+ OK,
+ DB_NOT_FOUND,
+ TABLE_NOT_FOUND,
+ FUNCTION_NOT_FOUND
+}
+
// RPC response for GetPartialCatalogObject.
struct TGetPartialCatalogObjectResponse {
// The status of the operation, OK if the operation was successful.
// Unset indicates "OK".
1: optional Status.TStatus status
- 2: optional i64 object_version_number
- 3: optional TPartialTableInfo table_info
- 4: optional TPartialDbInfo db_info
- 5: optional TPartialCatalogInfo catalog_info
+ // Catalog-specific error codes (eg if the object no longer exists).
+ 2: optional CatalogLookupStatus lookup_status = CatalogLookupStatus.OK
+
+ 3: optional i64 object_version_number
+ 4: optional TPartialTableInfo table_info
+ 5: optional TPartialDbInfo db_info
+ 6: optional TPartialCatalogInfo catalog_info
// Functions are small enough that we return them wholesale.
- 6: optional list<Types.TFunction> functions
+ 7: optional list<Types.TFunction> functions
}
http://git-wip-us.apache.org/repos/asf/impala/blob/94cfdac6/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
----------------------------------------------------------------------
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 2f5e9e4..a01de9a 100644
--- a/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
+++ b/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
@@ -49,6 +49,7 @@ import org.apache.impala.common.Reference;
import org.apache.impala.common.RuntimeEnv;
import org.apache.impala.service.BackendConfig;
import org.apache.impala.service.FeSupport;
+import org.apache.impala.thrift.CatalogLookupStatus;
import org.apache.impala.thrift.CatalogServiceConstants;
import org.apache.impala.thrift.TCatalog;
import org.apache.impala.thrift.TCatalogInfoSelector;
@@ -2094,7 +2095,9 @@ public class CatalogServiceCatalog extends Catalog {
/**
* Return a partial view of information about a given catalog object. This services
* the CatalogdMetaProvider running on impalads when they are configured in
- * "local-catalog" mode.
+ * "local-catalog" mode. If required objects are not present, for example, the database
+ * from which a table is requested, the types of the missing objects will be set in the
+ * response's lookup_status.
*/
public TGetPartialCatalogObjectResponse getPartialCatalogObject(
TGetPartialCatalogObjectRequest req) throws CatalogException {
@@ -2109,8 +2112,7 @@ public class CatalogServiceCatalog extends Catalog {
try {
Db db = getDb(dbDesc.getDb_name());
if (db == null) {
- throw new CatalogException(
- "Database not found: " + req.object_desc.getDb().getDb_name());
+ return createGetPartialCatalogObjectError(CatalogLookupStatus.DB_NOT_FOUND);
}
return db.getPartialInfo(req);
@@ -2119,11 +2121,15 @@ public class CatalogServiceCatalog extends Catalog {
}
case TABLE:
case VIEW: {
- Table table = getOrLoadTable(objectDesc.getTable().getDb_name(),
- objectDesc.getTable().getTbl_name());
+ Table table;
+ try {
+ table = getOrLoadTable(
+ objectDesc.getTable().getDb_name(), objectDesc.getTable().getTbl_name());
+ } catch (DatabaseNotFoundException e) {
+ return createGetPartialCatalogObjectError(CatalogLookupStatus.DB_NOT_FOUND);
+ }
if (table == null) {
- throw new CatalogException("Table not found: " +
- objectDesc.getTable().getTbl_name());
+ return createGetPartialCatalogObjectError(CatalogLookupStatus.TABLE_NOT_FOUND);
}
// TODO(todd): consider a read-write lock here.
table.getLock().lock();
@@ -2138,11 +2144,14 @@ public class CatalogServiceCatalog extends Catalog {
try {
Db db = getDb(objectDesc.fn.name.db_name);
if (db == null) {
- throw new CatalogException(
- "Database not found: " + objectDesc.fn.name.db_name);
+ return createGetPartialCatalogObjectError(CatalogLookupStatus.DB_NOT_FOUND);
}
List<Function> funcs = db.getFunctions(objectDesc.fn.name.function_name);
+ if (funcs.isEmpty()) {
+ return createGetPartialCatalogObjectError(
+ CatalogLookupStatus.FUNCTION_NOT_FOUND);
+ }
TGetPartialCatalogObjectResponse resp = new TGetPartialCatalogObjectResponse();
List<TFunction> thriftFuncs = Lists.newArrayListWithCapacity(funcs.size());
for (Function f : funcs) thriftFuncs.add(f.toThrift());
@@ -2158,6 +2167,13 @@ public class CatalogServiceCatalog extends Catalog {
}
}
+ private static TGetPartialCatalogObjectResponse createGetPartialCatalogObjectError(
+ CatalogLookupStatus status) {
+ TGetPartialCatalogObjectResponse resp = new TGetPartialCatalogObjectResponse();
+ resp.setLookup_status(status);
+ return resp;
+ }
+
/**
* Return a partial view of information about global parts of the catalog (eg
* the list of tables, etc).
http://git-wip-us.apache.org/repos/asf/impala/blob/94cfdac6/fe/src/main/java/org/apache/impala/catalog/local/CatalogdMetaProvider.java
----------------------------------------------------------------------
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 6888463..ca9a765 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
@@ -48,6 +48,7 @@ import org.apache.impala.common.InternalException;
import org.apache.impala.common.Pair;
import org.apache.impala.common.Reference;
import org.apache.impala.service.FeSupport;
+import org.apache.impala.thrift.CatalogLookupStatus;
import org.apache.impala.thrift.TBackendGflags;
import org.apache.impala.thrift.TCatalogInfoSelector;
import org.apache.impala.thrift.TCatalogObject;
@@ -88,6 +89,7 @@ import com.google.common.cache.Weigher;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
+import com.google.common.util.concurrent.UncheckedExecutionException;
import com.google.errorprone.annotations.Immutable;
import com.google.errorprone.annotations.concurrent.GuardedBy;
@@ -301,6 +303,24 @@ public class CatalogdMetaProvider implements MetaProvider {
throw new TException(resp.toString());
}
+ // If we get a "not found" response, then we assume that this was a case of an
+ // inconsistent cache. For example, we might have cached the list of tables within
+ // a database, but the table we're trying to load was dropped just prior to us
+ // trying to load it. In these cases, we need to invalidate whatever cache items
+ // might have led us to the dropped object and throw an exception so that we
+ // can retry with a reloaded cache.
+ switch (resp.lookup_status) {
+ case DB_NOT_FOUND:
+ case FUNCTION_NOT_FOUND:
+ case TABLE_NOT_FOUND:
+ invalidateCacheForObject(req.object_desc);
+ throw new InconsistentMetadataFetchException(
+ String.format("Fetching {} failed. Could not find {}",
+ req.object_desc.type.name(), resp.lookup_status.name()));
+ default: break;
+ }
+ Preconditions.checkState(resp.lookup_status == CatalogLookupStatus.OK);
+
// If we requested information about a particular version of an object, but
// got back a response for a different version, then we have a case of "read skew".
// For example, we may have fetched the partition list of a table, performed pruning,
@@ -311,11 +331,12 @@ public class CatalogdMetaProvider implements MetaProvider {
resp.isSetObject_version_number() &&
req.object_desc.catalog_version != resp.object_version_number) {
invalidateCacheForObject(req.object_desc);
- throw new InconsistentMetadataFetchException(String.format(
- "Catalog object %s changed version from %s to %s while fetching metadata",
- req.object_desc.toString(),
- req.object_desc.catalog_version,
- resp.object_version_number));
+ LOG.warn("Catalog object {} changed version from {} to {} while fetching metadata",
+ req.object_desc.toString(), req.object_desc.catalog_version,
+ resp.object_version_number);
+ throw new InconsistentMetadataFetchException(
+ String.format("Catalog object %s changed version between accesses.",
+ req.object_desc.toString()));
}
return resp;
}
@@ -337,7 +358,7 @@ public class CatalogdMetaProvider implements MetaProvider {
return loadCallable.call();
}
});
- } catch (ExecutionException e) {
+ } catch (ExecutionException | UncheckedExecutionException e) {
Throwables.propagateIfPossible(e.getCause(), TException.class);
throw new RuntimeException(e);
} finally {
@@ -522,37 +543,27 @@ public class CatalogdMetaProvider implements MetaProvider {
@Override
public List<PartitionRef> loadPartitionList(final TableMetaRef table)
throws TException {
- final Reference<Boolean> hitCache = new Reference<>(true);
- try {
- PartitionListCacheKey key = new PartitionListCacheKey((TableMetaRefImpl)table);
- return (List<PartitionRef>) cache_.get(key,
- new Callable<List<PartitionRef>>() {
- /** Called to load cache for cache misses */
- @Override
- public List<PartitionRef> call() throws Exception {
- hitCache.setRef(false);
- TGetPartialCatalogObjectRequest req = newReqForTable(table);
- req.table_info_selector.want_partition_names = true;
- TGetPartialCatalogObjectResponse resp = sendRequest(req);
- checkResponse(resp.table_info != null && resp.table_info.partitions != null,
- req, "missing partition list result");
- List<PartitionRef> partitionRefs = Lists.newArrayListWithCapacity(
- resp.table_info.partitions.size());
- for (TPartialPartitionInfo p: resp.table_info.partitions) {
- checkResponse(p.isSetId(), req,
- "response missing partition IDs for partition %s", p);
- partitionRefs.add(new PartitionRefImpl(p));
- }
- return partitionRefs;
+ PartitionListCacheKey key = new PartitionListCacheKey((TableMetaRefImpl) table);
+ return (List<PartitionRef>) loadWithCaching(
+ "partition list for " + table, key, new Callable<List<PartitionRef>>() {
+ /** Called to load cache for cache misses */
+ @Override
+ public List<PartitionRef> call() throws Exception {
+ TGetPartialCatalogObjectRequest req = newReqForTable(table);
+ req.table_info_selector.want_partition_names = true;
+ TGetPartialCatalogObjectResponse resp = sendRequest(req);
+ checkResponse(resp.table_info != null && resp.table_info.partitions != null,
+ req, "missing partition list result");
+ List<PartitionRef> partitionRefs =
+ Lists.newArrayListWithCapacity(resp.table_info.partitions.size());
+ for (TPartialPartitionInfo p : resp.table_info.partitions) {
+ checkResponse(
+ p.isSetId(), req, "response missing partition IDs for partition %s", p);
+ partitionRefs.add(new PartitionRefImpl(p));
}
- });
- } catch (ExecutionException e) {
- Throwables.propagateIfPossible(e.getCause(), TException.class);
- throw new RuntimeException(e);
- } finally {
- LOG.trace("Request for partition list of {}: {}", table,
- hitCache.getRef() ? "hit":"miss");
- }
+ return partitionRefs;
+ }
+ });
}
@Override
@@ -712,19 +723,14 @@ public class CatalogdMetaProvider implements MetaProvider {
@Override
public String loadNullPartitionKeyValue() throws MetaException, TException {
- try {
- return (String) cache_.get(NULL_PARTITION_KEY_VALUE_CACHE_KEY,
- new Callable<String>() {
- /** Called to load cache for cache misses */
- @Override
- public String call() throws Exception {
- return directProvider_.loadNullPartitionKeyValue();
- }
- });
- } catch (ExecutionException e) {
- Throwables.propagateIfPossible(e.getCause(), TException.class);
- throw new RuntimeException(e);
- }
+ return (String) loadWithCaching("null partition key value",
+ NULL_PARTITION_KEY_VALUE_CACHE_KEY, new Callable<String>() {
+ /** Called to load cache for cache misses */
+ @Override
+ public String call() throws Exception {
+ return directProvider_.loadNullPartitionKeyValue();
+ }
+ });
}
@Override
@@ -961,7 +967,8 @@ public class CatalogdMetaProvider implements MetaProvider {
/**
* Invalidate items from the cache in response to seeing an updated catalog object
- * from the catalogd.
+ * from the catalogd or getting an error response from another request that indicates
+ * that the object has been removed.
*/
@VisibleForTesting
void invalidateCacheForObject(TCatalogObject obj) {
http://git-wip-us.apache.org/repos/asf/impala/blob/94cfdac6/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 783f972..22a1b70 100644
--- a/fe/src/main/java/org/apache/impala/service/Frontend.java
+++ b/fe/src/main/java/org/apache/impala/service/Frontend.java
@@ -88,6 +88,7 @@ import org.apache.impala.catalog.Function;
import org.apache.impala.catalog.ImpaladCatalog;
import org.apache.impala.catalog.ImpaladTableUsageTracker;
import org.apache.impala.catalog.Type;
+import org.apache.impala.catalog.local.InconsistentMetadataFetchException;
import org.apache.impala.common.AnalysisException;
import org.apache.impala.common.FileSystemUtil;
import org.apache.impala.common.ImpalaException;
@@ -152,6 +153,7 @@ import com.google.common.base.Preconditions;
import com.google.common.base.Predicates;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
+import com.google.common.util.concurrent.Uninterruptibles;
/**
* Frontend API for the impalad process.
@@ -168,6 +170,9 @@ public class Frontend {
// TODO: Make the reload interval configurable.
private static final int AUTHORIZATION_POLICY_RELOAD_INTERVAL_SECS = 5 * 60;
+ // Maximum number of times to retry a query if it fails due to inconsistent metadata.
+ private static final int INCONSISTENT_METADATA_NUM_RETRIES = 10;
+
private final FeCatalogManager catalogManager_;
private final AuthorizationConfig authzConfig_;
/**
@@ -1041,9 +1046,6 @@ public class Frontend {
*/
public TExecRequest createExecRequest(TQueryCtx queryCtx, StringBuilder explainString)
throws ImpalaException {
- // TODO(todd): wrap the planning in a retry loop which catches
- // InconsistentMetadataFetchException.
-
// Timeline of important events in the planning process, used for debugging
// and profiling.
EventSequence timeline = new EventSequence("Query Compilation");
@@ -1057,6 +1059,31 @@ public class Frontend {
StringBuilder explainString) throws ImpalaException {
LOG.info("Analyzing query: " + queryCtx.client_request.stmt);
+ int attempt = 0;
+ while (true) {
+ try {
+ return doCreateExecRequest(queryCtx, timeline, explainString);
+ } catch (InconsistentMetadataFetchException e) {
+ if (attempt++ == INCONSISTENT_METADATA_NUM_RETRIES) {
+ throw e;
+ }
+ if (attempt > 1) {
+ // Back-off a bit on later retries.
+ Uninterruptibles.sleepUninterruptibly(200 * attempt, TimeUnit.MILLISECONDS);
+ }
+ timeline.markEvent(
+ String.format("Retrying query planning due to inconsistent metadata "
+ + "fetch, attempt %s of %s: ",
+ attempt, INCONSISTENT_METADATA_NUM_RETRIES)
+ + e.getMessage());
+ LOG.warn("Retrying plan of query {}: {} (retry #{})",
+ queryCtx.client_request.stmt, e.getMessage(), attempt);
+ }
+ }
+ }
+
+ private TExecRequest doCreateExecRequest(TQueryCtx queryCtx, EventSequence timeline,
+ StringBuilder explainString) throws ImpalaException {
// Parse stmt and collect/load metadata to populate a stmt-local table cache
StatementBase stmt = parse(queryCtx.client_request.stmt);
StmtMetadataLoader metadataLoader =
http://git-wip-us.apache.org/repos/asf/impala/blob/94cfdac6/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 634adca..5eaf4ac 100644
--- a/tests/custom_cluster/test_local_catalog.py
+++ b/tests/custom_cluster/test_local_catalog.py
@@ -18,10 +18,13 @@
# Test behaviors specific to --use_local_catalog being enabled.
import pytest
+import random
+import threading
import time
from tests.common.custom_cluster_test_suite import CustomClusterTestSuite
+RETRY_PROFILE_MSG = 'Retrying query planning due to inconsistent metadata'
class TestCompactCatalogUpdates(CustomClusterTestSuite):
@pytest.mark.execute_serially
@@ -104,7 +107,7 @@ class TestCompactCatalogUpdates(CustomClusterTestSuite):
finally:
client.close()
-
+ @pytest.mark.execute_serially
@CustomClusterTestSuite.with_args(
impalad_args="--use_local_catalog=true",
catalogd_args="--catalog_topic_mode=minimal")
@@ -146,3 +149,64 @@ class TestCompactCatalogUpdates(CustomClusterTestSuite):
finally:
client.close()
+
+ @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,
+ the query planner retries the query.
+ """
+ try:
+ impalad1 = self.cluster.impalads[0]
+ impalad2 = self.cluster.impalads[1]
+ client1 = impalad1.service.create_beeswax_client()
+ client2 = impalad2.service.create_beeswax_client()
+
+ # Create a view in client 1, cache the table list including that view in
+ # client 2, and then drop it in client 1. While we've still cached the
+ # table list, try to describe the view from client 2 -- it should fail
+ # with the normal error message even though it had the inconsistent cache.
+ view = "%s.my_view" % unique_database
+ self.execute_query_expect_success(client1, "create view %s as select 1" % view)
+ self.execute_query_expect_success(client2, "show tables")
+ self.execute_query_expect_success(client1, "drop view %s" % view)
+ err = self.execute_query_expect_failure(client2, "describe %s" % view)
+ assert "Could not resolve path" in str(err)
+
+ # Run a mix of concurrent REFRESH and queries against different subsets
+ # of partitions. This causes partial views of the table to get cached,
+ # and then as the new partitions are loaded, we detect the version skew
+ # and issue re-plans. We run the concurrent workload until the profile
+ # 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.
+ replans_seen = [0]
+ replans_seen_lock = threading.Lock()
+
+ def stress_thread(client):
+ while replans_seen[0] == 0:
+ # TODO(todd) EXPLAIN queries don't currently yield a profile, so
+ # we have to actually run a COUNT query.
+ q = random.choice([
+ 'refresh functional.alltypes',
+ 'select count(*) from functional.alltypes where month=4',
+ 'select count(*) from functional.alltypes where month=5'])
+ ret = self.execute_query_expect_success(client, q)
+ if RETRY_PROFILE_MSG in ret.runtime_profile:
+ with replans_seen_lock:
+ replans_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(30)
+ assert replans_seen[0] > 0, "Did not trigger any re-plans"
+
+ finally:
+ client1.close()
+ client2.close()