diff --git a/be/src/common/global-flags.cc b/be/src/common/global-flags.cc
index 451da40..40841b5 100644
--- a/be/src/common/global-flags.cc
+++ b/be/src/common/global-flags.cc
@@ -300,6 +300,14 @@
     "may result in false positives"
     "This overrides KRPC's --rpc_duration_too_long_ms setting.");
 
+DEFINE_int32(num_check_authorization_threads, 1,
+    "The number of threads used to check authorization for the user when executing show "
+    "tables/databases. This configuration is applicable only when authorization is "
+    "enabled. A value of 1 disables multi-threaded execution for checking authorization."
+    "However, a small value of larger than 1 may limit the parallism of FE requests when "
+    "checking authorization with a high concurrency. The value must be in the range of "
+    "1 to 128.");
+
 // ++========================++
 // || Startup flag graveyard ||
 // ++========================++
diff --git a/be/src/util/backend-gflag-util.cc b/be/src/util/backend-gflag-util.cc
index 6906e1d..467873b 100644
--- a/be/src/util/backend-gflag-util.cc
+++ b/be/src/util/backend-gflag-util.cc
@@ -83,6 +83,7 @@
 DECLARE_string(blacklisted_tables);
 DECLARE_string(min_privilege_set_for_show_stmts);
 DECLARE_int32(num_expected_executors);
+DECLARE_int32(num_check_authorization_threads);
 
 namespace impala {
 
@@ -169,6 +170,7 @@
   cfg.__set_blacklisted_tables(FLAGS_blacklisted_tables);
   cfg.__set_min_privilege_set_for_show_stmts(FLAGS_min_privilege_set_for_show_stmts);
   cfg.__set_num_expected_executors(FLAGS_num_expected_executors);
+  cfg.__set_num_check_authorization_threads(FLAGS_num_check_authorization_threads);
   RETURN_IF_ERROR(SerializeThriftMsg(jni_env, &cfg, cfg_bytes));
   return Status::OK();
 }
diff --git a/common/thrift/BackendGflags.thrift b/common/thrift/BackendGflags.thrift
index 3489793..4762654 100644
--- a/common/thrift/BackendGflags.thrift
+++ b/common/thrift/BackendGflags.thrift
@@ -147,4 +147,6 @@
   61: required bool mt_dop_auto_fallback
 
   62: required i32 num_expected_executors
+
+  63: required i32 num_check_authorization_threads
 }
diff --git a/fe/src/main/java/org/apache/impala/service/BackendConfig.java b/fe/src/main/java/org/apache/impala/service/BackendConfig.java
index 47c4660..d3abf78 100644
--- a/fe/src/main/java/org/apache/impala/service/BackendConfig.java
+++ b/fe/src/main/java/org/apache/impala/service/BackendConfig.java
@@ -197,6 +197,10 @@
     return backendCfg_.min_privilege_set_for_show_stmts;
   }
 
+  public int getNumCheckAuthorizationThreads() {
+    return backendCfg_.num_check_authorization_threads;
+  }
+
   // Inits the auth_to_local configuration in the static KerberosName class.
   private static void initAuthToLocal() {
     // If auth_to_local is enabled, we read the configuration hadoop.security.auth_to_local
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 5c36f37..f01d787 100644
--- a/fe/src/main/java/org/apache/impala/service/Frontend.java
+++ b/fe/src/main/java/org/apache/impala/service/Frontend.java
@@ -33,6 +33,9 @@
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ExecutionException;
 
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
@@ -172,6 +175,7 @@
 import com.google.common.base.Strings;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
+import com.google.common.util.concurrent.MoreExecutors;
 import com.google.common.util.concurrent.Uninterruptibles;
 
 /**
@@ -193,6 +197,10 @@
   private static final int INCONSISTENT_METADATA_NUM_RETRIES =
       BackendConfig.INSTANCE.getLocalCatalogMaxFetchRetries();
 
+  // Maximum number of threads used to check authorization for the user when executing
+  // show tables/databases.
+  private static final int MAX_CHECK_AUTHORIZATION_POOL_SIZE = 128;
+
   /**
    * Plan-time context that allows capturing various artifacts created
    * during the process.
@@ -271,6 +279,8 @@
 
   private final TransactionKeepalive transactionKeepalive_;
 
+  private static ExecutorService checkAuthorizationPool_;
+
   public Frontend(AuthorizationFactory authzFactory) throws ImpalaException {
     this(authzFactory, FeCatalogManager.createFromBackendConfig());
   }
@@ -294,6 +304,15 @@
     if (authzConfig.isEnabled()) {
       authzChecker_.set(authzFactory.newAuthorizationChecker(
           getCatalog().getAuthPolicy()));
+      int numThreads = BackendConfig.INSTANCE.getNumCheckAuthorizationThreads();
+      Preconditions.checkState(numThreads > 0
+        && numThreads <= MAX_CHECK_AUTHORIZATION_POOL_SIZE);
+      if (numThreads == 1) {
+        checkAuthorizationPool_ = MoreExecutors.sameThreadExecutor();
+      } else {
+        LOG.info("Using a thread pool of size {} for authorization", numThreads);
+        checkAuthorizationPool_ = Executors.newFixedThreadPool(numThreads);
+      }
     } else {
       authzChecker_.set(authzFactory.newAuthorizationChecker());
     }
@@ -776,6 +795,31 @@
   }
 
   /**
+   * A Callable wrapper used for checking authorization to tables/databases.
+   */
+  private class CheckAuthorization implements Callable<Boolean> {
+    private final String dbName_;
+    private final String tblName_;
+    private final String owner_;
+    private final User user_;
+
+    public CheckAuthorization(String dbName, String tblName, String owner, User user) {
+      // dbName and user cannot be null, tblName and owner can be null.
+      Preconditions.checkNotNull(dbName);
+      Preconditions.checkNotNull(user);
+      dbName_ = dbName;
+      tblName_ = tblName;
+      owner_ = owner;
+      user_ = user;
+    }
+
+    @Override
+    public Boolean call() throws Exception {
+      return new Boolean(isAccessibleToUser(dbName_, tblName_, owner_, user_));
+    }
+  }
+
+  /**
    * Returns all tables in database 'dbName' that match the pattern of 'matcher' and are
    * accessible to 'user'.
    */
@@ -792,11 +836,40 @@
     }
   }
 
+  /**
+   * This method filters out elements from the given list based on the the results
+   * of the pendingCheckTasks.
+   */
+  private void filterUnaccessibleElements(List<Future<Boolean>> pendingCheckTasks,
+    List<?> checkList) throws InternalException {
+    int failedCheckTasks = 0;
+    int index = 0;
+    Iterator<?> iter = checkList.iterator();
+
+    Preconditions.checkState(checkList.size() == pendingCheckTasks.size());
+    while (iter.hasNext()) {
+      iter.next();
+      try {
+        if (!pendingCheckTasks.get(index).get()) iter.remove();
+        index++;
+      } catch (ExecutionException | InterruptedException e) {
+        failedCheckTasks++;
+        LOG.error("Encountered an error checking access", e);
+        break;
+      }
+    }
+
+    if (failedCheckTasks > 0)
+      throw new InternalException("Failed to check access." +
+          "Check the server log for more details.");
+  }
+
   private List<String> doGetTableNames(String dbName, PatternMatcher matcher,
       User user) throws ImpalaException {
     FeCatalog catalog = getCatalog();
     List<String> tblNames = catalog.getTableNames(dbName, matcher);
     if (authzFactory_.getAuthorizationConfig().isEnabled()) {
+      List<Future<Boolean>> pendingCheckTasks = Lists.newArrayList();
       Iterator<String> iter = tblNames.iterator();
       while (iter.hasNext()) {
         String tblName = iter.next();
@@ -811,18 +884,15 @@
         String tableOwner = table.getOwnerUser();
         if (tableOwner == null) {
           LOG.info("Table {} not yet loaded, ignoring it in table listing.",
-              dbName + "." + tblName);
+            dbName + "." + tblName);
         }
-        Set<PrivilegeRequest> requests = new PrivilegeRequestBuilder(
-            authzFactory_.getAuthorizableFactory())
-            .anyOf(minPrivilegeSetForShowStmts_)
-            .onAnyColumn(dbName, tblName, tableOwner)
-            .buildSet();
-        if (!authzChecker_.get().hasAnyAccess(user, requests)) {
-          iter.remove();
-        }
+        pendingCheckTasks.add(checkAuthorizationPool_.submit(
+            new CheckAuthorization(dbName, tblName, tableOwner, user)));
       }
+
+      filterUnaccessibleElements(pendingCheckTasks, tblNames);
     }
+
     return tblNames;
   }
 
@@ -944,29 +1014,43 @@
     // have permissions on.
     if (authzFactory_.getAuthorizationConfig().isEnabled()) {
       Iterator<? extends FeDb> iter = dbs.iterator();
+      List<Future<Boolean>> pendingCheckTasks = Lists.newArrayList();
       while (iter.hasNext()) {
         FeDb db = iter.next();
-        if (!isAccessibleToUser(db, user)) iter.remove();
+        pendingCheckTasks.add(checkAuthorizationPool_.submit(
+            new CheckAuthorization(db.getName(), null, db.getOwnerUser(), user)));
       }
+
+      filterUnaccessibleElements(pendingCheckTasks, dbs);
     }
+
     return dbs;
   }
 
   /**
-   * Check whether database is accessible to given user.
+   * Check whether table/database is accessible to given user.
    */
-  private boolean isAccessibleToUser(FeDb db, User user)
-      throws InternalException {
-    if (db.getName().toLowerCase().equals(Catalog.DEFAULT_DB.toLowerCase())) {
+  private boolean isAccessibleToUser(String dbName, String tblName,
+      String owner, User user) throws InternalException {
+    Preconditions.checkNotNull(dbName);
+    if (tblName == null &&
+        dbName.toLowerCase().equals(Catalog.DEFAULT_DB.toLowerCase())) {
       // Default DB should always be shown.
       return true;
     }
-    Set<PrivilegeRequest> requests = new PrivilegeRequestBuilder(
-        authzFactory_.getAuthorizableFactory())
-        .anyOf(minPrivilegeSetForShowStmts_)
-        .onAnyColumn(db.getName(), db.getOwnerUser())
-        .buildSet();
-    return authzChecker_.get().hasAnyAccess(user, requests);
+
+    PrivilegeRequestBuilder builder = new PrivilegeRequestBuilder(
+      authzFactory_.getAuthorizableFactory())
+      .anyOf(minPrivilegeSetForShowStmts_);
+    if (tblName == null) {
+      // Check database
+      builder = builder.onAnyColumn(dbName, owner);
+    } else {
+      // Check table
+      builder = builder.onAnyColumn(dbName, tblName, owner);
+    }
+
+    return authzChecker_.get().hasAnyAccess(user, builder.buildSet());
   }
 
   /**
diff --git a/tests/authorization/test_authorization.py b/tests/authorization/test_authorization.py
index 885d184..7f57ddb 100644
--- a/tests/authorization/test_authorization.py
+++ b/tests/authorization/test_authorization.py
@@ -22,6 +22,7 @@
 import tempfile
 import grp
 import re
+import random
 import sys
 import subprocess
 import urllib
@@ -651,3 +652,26 @@
                   "--ranger_app_id=impala --authorization_provider=ranger")
   def test_ranger_show_stmts_with_any(self, unique_name):
     self._test_ranger_show_stmts_helper(unique_name, PRIVILEGES)
+
+  @pytest.mark.execute_serially
+  @CustomClusterTestSuite.with_args(
+    impalad_args="--server-name=server1 --ranger_service_type=hive "
+                 "--ranger_app_id=impala --authorization_provider=ranger "
+                 "--num_check_authorization_threads=%d" % (random.randint(2, 128)),
+    catalogd_args="--server-name=server1 --ranger_service_type=hive "
+                  "--ranger_app_id=impala --authorization_provider=ranger")
+  def test_num_check_authorization_threads_with_ranger(self, unique_name):
+    self._test_ranger_show_stmts_helper(unique_name, PRIVILEGES)
+
+  @pytest.mark.execute_serially
+  @CustomClusterTestSuite.with_args(
+    impalad_args="--server_name=server1 --sentry_config=%s "
+                 "--authorized_proxy_user_config=%s=* "
+                 "--num_check_authorization_threads=%d" %
+                 (SENTRY_CONFIG_FILE, getuser(), random.randint(2, 128)),
+    catalogd_args="--sentry_config={0}".format(SENTRY_CONFIG_FILE),
+    sentry_config=SENTRY_CONFIG_FILE_OO,  # Enable Sentry Object Ownership
+    sentry_log_dir="{0}/test_num_check_authorization_threads_with_sentry"
+                   .format(SENTRY_BASE_LOG_DIR))
+  def test_num_check_authorization_threads_with_sentry(self, unique_role, unique_name):
+    self._test_sentry_show_stmts_helper(unique_role, unique_name, PRIVILEGES)