You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by ta...@apache.org on 2019/02/19 18:28:57 UTC
[impala] branch 2.x updated (e660f11 -> 800ac7d)
This is an automated email from the ASF dual-hosted git repository.
tarmstrong pushed a change to branch 2.x
in repository https://gitbox.apache.org/repos/asf/impala.git.
from e660f11 IMPALA-7111: avoid use of boost::split in CheckPluginEnabled
new 31c3640 IMPALA-7174: fix test_cancellation for RELEASE builds
new c561fcc IMPALA-7135. Skeleton implementation of LocalCatalog
new b969566 [DOCS] Fixed the DITA concept ID for the new impala_disable_codegen_rows_threshold
new 08d634f IMPALA-7179: allow_multiple_scratch_dirs_per_device=true by default
new a8d2f2f IMPALA-7130: impala-shell -b / --kerberos_host_fqdn flag overrides value passed in via -i / --impalad
new 44a3594 [DOCS] Fixed a typo for missing 'not'
new 9c50750 IMPALA-7050: [DOCS] Document the max serialized incremental stat size setting
new 800ac7d IMPALA-7137. Support configuring Frontend to use LocalCatalog
The 8 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
be/src/runtime/exec-env.cc | 4 +
be/src/runtime/tmp-file-mgr.cc | 2 +-
be/src/scheduling/admission-controller.cc | 4 -
be/src/service/client-request-state.cc | 6 -
be/src/service/impala-server.cc | 3 +-
be/src/util/backend-gflag-util.cc | 2 +
be/src/util/debug-util.cc | 4 +-
be/src/util/debug-util.h | 15 +-
common/thrift/BackendGflags.thrift | 2 +
.../impala_disable_codegen_rows_threshold.xml | 20 +-
docs/topics/impala_perf_cookbook.xml | 5 +-
docs/topics/impala_perf_stats.xml | 45 +++++
.../java/org/apache/impala/catalog/Catalog.java | 8 +-
.../org/apache/impala/catalog/CatalogObject.java | 3 +-
.../main/java/org/apache/impala/catalog/FeDb.java | 3 +-
.../catalog/{FeDataSource.java => HasName.java} | 16 +-
.../impala/catalog/local/DirectMetaProvider.java | 83 ++++++++
.../apache/impala/catalog/local/LocalCatalog.java | 217 +++++++++++++++++++++
.../local/LocalCatalogException.java} | 23 +--
.../org/apache/impala/catalog/local/LocalDb.java | 175 +++++++++++++++++
.../apache/impala/catalog/local/LocalTable.java | 160 +++++++++++++++
.../apache/impala/catalog/local/MetaProvider.java | 48 +++++
.../impala/service/DescribeResultFactory.java | 12 +-
.../apache/impala/service/FeCatalogManager.java | 157 +++++++++++++++
.../java/org/apache/impala/service/Frontend.java | 104 +++++-----
.../org/apache/impala/service/JniFrontend.java | 2 +-
.../java/org/apache/impala/service/MetadataOp.java | 8 +-
.../impala/catalog/local/LocalCatalogTest.java | 76 ++++++++
shell/impala_client.py | 23 ++-
tests/custom_cluster/test_admission_controller.py | 6 +-
tests/custom_cluster/test_scratch_disk.py | 3 +-
tests/shell/test_shell_commandline.py | 35 +++-
32 files changed, 1137 insertions(+), 137 deletions(-)
copy fe/src/main/java/org/apache/impala/catalog/{FeDataSource.java => HasName.java} (81%)
create mode 100644 fe/src/main/java/org/apache/impala/catalog/local/DirectMetaProvider.java
create mode 100644 fe/src/main/java/org/apache/impala/catalog/local/LocalCatalog.java
copy fe/src/main/java/org/apache/impala/{common/ImpalaException.java => catalog/local/LocalCatalogException.java} (61%)
create mode 100644 fe/src/main/java/org/apache/impala/catalog/local/LocalDb.java
create mode 100644 fe/src/main/java/org/apache/impala/catalog/local/LocalTable.java
create mode 100644 fe/src/main/java/org/apache/impala/catalog/local/MetaProvider.java
create mode 100644 fe/src/main/java/org/apache/impala/service/FeCatalogManager.java
create mode 100644 fe/src/test/java/org/apache/impala/catalog/local/LocalCatalogTest.java
[impala] 06/08: [DOCS] Fixed a typo for missing 'not'
Posted by ta...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tarmstrong pushed a commit to branch 2.x
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 44a3594c325e2fe0d4a4ed8801fbb26061a7b61e
Author: Alex Rodoni <ar...@cloudera.com>
AuthorDate: Fri Jun 8 12:47:09 2018 -0700
[DOCS] Fixed a typo for missing 'not'
Change-Id: I8342bcb47d4a9aa422e234e488dd1dfbdc1694d4
Reviewed-on: http://gerrit.cloudera.org:8080/10657
Reviewed-by: Balazs Jeszenszky <je...@gmail.com>
Reviewed-by: Alex Rodoni <ar...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
docs/topics/impala_perf_cookbook.xml | 5 ++---
1 file changed, 2 insertions(+), 3 deletions(-)
diff --git a/docs/topics/impala_perf_cookbook.xml b/docs/topics/impala_perf_cookbook.xml
index 29b5a13..8a5e603 100644
--- a/docs/topics/impala_perf_cookbook.xml
+++ b/docs/topics/impala_perf_cookbook.xml
@@ -340,9 +340,8 @@ under the License.
</li>
<li>
- Do not compress the table data. Compressing the table data will
- likely cause the data to span more nodes and eliminate skew caused by
- compression.
+ Do not compress the table data. The uncompressed table data spans more
+ nodes and eliminates skew caused by compression.
</li>
<li>
[impala] 03/08: [DOCS] Fixed the DITA concept ID for the new
impala_disable_codegen_rows_threshold
Posted by ta...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tarmstrong pushed a commit to branch 2.x
in repository https://gitbox.apache.org/repos/asf/impala.git
commit b96956642497b8a5017eaab5d0cb37f5af26d3aa
Author: Alex Rodoni <ar...@cloudera.com>
AuthorDate: Fri Jun 15 12:37:22 2018 -0700
[DOCS] Fixed the DITA concept ID for the new impala_disable_codegen_rows_threshold
Change-Id: I8ecbf93749d5eb8da1d5c873fd7503f5bb2c7d0f
Reviewed-on: http://gerrit.cloudera.org:8080/10733
Reviewed-by: Alex Rodoni <ar...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
.../topics/impala_disable_codegen_rows_threshold.xml | 20 +++++++++-----------
1 file changed, 9 insertions(+), 11 deletions(-)
diff --git a/docs/topics/impala_disable_codegen_rows_threshold.xml b/docs/topics/impala_disable_codegen_rows_threshold.xml
index b16a691..438db3d 100644
--- a/docs/topics/impala_disable_codegen_rows_threshold.xml
+++ b/docs/topics/impala_disable_codegen_rows_threshold.xml
@@ -18,7 +18,7 @@ specific language governing permissions and limitations
under the License.
-->
<!DOCTYPE concept PUBLIC "-//OASIS//DTD DITA Concept//EN" "concept.dtd">
-<concept rev="2.0.0" id="exec_single_node_rows_threshold">
+<concept rev="2.10.0" id="disable_codegen_rows_threshold">
<title>DISABLE_CODEGEN_ROWS_THRESHOLD Query Option (<keyword keyref="impala210_full"/> or higher only)</title>
<titlealts audience="PDF"><navtitle>DISABLE_CODEGEN_ROWS_THRESHOLD</navtitle></titlealts>
@@ -35,16 +35,14 @@ under the License.
<conbody>
- <p rev="2.0.0">
- <indexterm audience="hidden">DISABLE_CODEGEN_ROWS_THRESHOLD query option</indexterm>
- This setting controls the cutoff point (in terms of number of rows processed per Impala daemon) below which
- Impala disables native code generation for the whole query.
-
- Native code generation is very beneficial for queries that process many rows because
- it reduces the time taken to process of each row. However, generating the native code
- adds latency to query startup. Therefore, automatically disabling codegen for
- queries that process relatively small amounts of data can improve query response time.
- </p>
+ <p rev="2.0.0"> This setting controls the cutoff point (in terms of number
+ of rows processed per Impala daemon) below which Impala disables native
+ code generation for the whole query. Native code generation is very
+ beneficial for queries that process many rows because it reduces the time
+ taken to process of each row. However, generating the native code adds
+ latency to query startup. Therefore, automatically disabling codegen for
+ queries that process relatively small amounts of data can improve query
+ response time. </p>
<p conref="../shared/impala_common.xml#common/syntax_blurb"/>
[impala] 04/08: IMPALA-7179:
allow_multiple_scratch_dirs_per_device=true by default
Posted by ta...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tarmstrong pushed a commit to branch 2.x
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 08d634ffb62609175d7113efbf7d0352672adf13
Author: Tim Armstrong <ta...@cloudera.com>
AuthorDate: Fri Jun 15 11:32:07 2018 -0700
IMPALA-7179: allow_multiple_scratch_dirs_per_device=true by default
The previous default was often confusing to users of Impala. It is
simpler to do exactly what is asked instead of trying to fix bad
configurations automatically.
Testing:
Ran core tests.
Change-Id: I23394c9949ae4cd0a21d7bb25551371b3198e76c
Reviewed-on: http://gerrit.cloudera.org:8080/10736
Reviewed-by: anujphadke <ap...@cloudera.com>
Reviewed-by: Dan Hecht <dh...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/runtime/tmp-file-mgr.cc | 2 +-
tests/custom_cluster/test_scratch_disk.py | 3 ++-
2 files changed, 3 insertions(+), 2 deletions(-)
diff --git a/be/src/runtime/tmp-file-mgr.cc b/be/src/runtime/tmp-file-mgr.cc
index a16775a..d6d5a51 100644
--- a/be/src/runtime/tmp-file-mgr.cc
+++ b/be/src/runtime/tmp-file-mgr.cc
@@ -42,7 +42,7 @@ DEFINE_bool(disk_spill_encryption, true,
"Set this to encrypt and perform an integrity "
"check on all data spilled to disk during a query");
DEFINE_string(scratch_dirs, "/tmp", "Writable scratch directories");
-DEFINE_bool(allow_multiple_scratch_dirs_per_device, false,
+DEFINE_bool(allow_multiple_scratch_dirs_per_device, true,
"If false and --scratch_dirs contains multiple directories on the same device, "
"then only the first writable directory is used");
diff --git a/tests/custom_cluster/test_scratch_disk.py b/tests/custom_cluster/test_scratch_disk.py
index 65bde66..c50ba24 100644
--- a/tests/custom_cluster/test_scratch_disk.py
+++ b/tests/custom_cluster/test_scratch_disk.py
@@ -82,7 +82,8 @@ class TestScratchDir(CustomClusterTestSuite):
scratch because all directories are on same disk."""
normal_dirs = self.generate_dirs(5)
self._start_impala_cluster([
- '--impalad_args="-logbuflevel=-1 -scratch_dirs={0}"'.format(','.join(normal_dirs))])
+ '--impalad_args="-logbuflevel=-1 -scratch_dirs={0}"'.format(','.join(normal_dirs)),
+ '--impalad_args=--allow_multiple_scratch_dirs_per_device=false'])
self.assert_impalad_log_contains("INFO", "Using scratch directory ",
expected_count=1)
exec_option = vector.get_value('exec_option')
[impala] 07/08: IMPALA-7050: [DOCS] Document the max serialized
incremental stat size setting
Posted by ta...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tarmstrong pushed a commit to branch 2.x
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 9c50750295668ec2c7ae192db681ec9b036f3581
Author: Alex Rodoni <ar...@cloudera.com>
AuthorDate: Fri May 18 17:53:54 2018 -0700
IMPALA-7050: [DOCS] Document the max serialized incremental stat size setting
Change-Id: Ifa80325f0008d42a9cc8178e7c144fc2b49d7d4e
Reviewed-on: http://gerrit.cloudera.org:8080/10457
Reviewed-by: Balazs Jeszenszky <je...@gmail.com>
Reviewed-by: Alex Rodoni <ar...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
docs/topics/impala_perf_stats.xml | 45 +++++++++++++++++++++++++++++++++++++++
1 file changed, 45 insertions(+)
diff --git a/docs/topics/impala_perf_stats.xml b/docs/topics/impala_perf_stats.xml
index dab2eb8..5e5d487 100644
--- a/docs/topics/impala_perf_stats.xml
+++ b/docs/topics/impala_perf_stats.xml
@@ -702,6 +702,51 @@ show column stats year_month_day;
</p>
</conbody>
+ <concept id="inc_stats_size_limit_bytes">
+ <title>Maximum Serialized Stats Size</title>
+ <conbody>
+ <p>
+ When executing <codeph>COMPUTE INCREMENTAL STATS</codeph> on
+ very large tables, use the configuration setting
+ <codeph>inc_stats_size_limit_bytes</codeph> to prevent Impala from
+ running out of memory while updating table metadata. If this limit
+ is reached, Impala will stop loading the table and return an error.
+ The error serves as an indication that <codeph>COMPUTE INCREMENTAL
+ STATS</codeph> should not be used on the particular table.
+ Consider spitting the table and using regular <codeph>COMPUTE
+ STATS</codeph> ]if possible.
+ </p>
+
+ <p>
+ The <codeph>inc_stats_size_limit_bytes</codeph> limit is set as a
+ safety check, to prevent Impala from hitting the maximum limit for
+ the table metadata. Note that this limit is only one part of the
+ entire table's metadata all of which together must be below 2 GB.
+ </p>
+
+ <p>
+ The default value for <codeph>inc_stats_size_limit_bytes</codeph>
+ is 20971520, 200 MB.
+ </p>
+
+ <p>
+ To change the <codeph>inc_stats_size_limit_bytes</codeph> value,
+ restart <codeph>impalad</codeph> and <codeph>catalogd</codeph> with
+ the new value specified in bytes, for example, 1048576000 for 1 GB.
+ See <xref
+ href="../../../../cloudera-docs/docs/topics/impala_config_options.xml#config_options"
+ /> for the steps to change the option and restart Impala
+ daemons.
+ </p>
+
+ <note type="attention">
+ The <codeph>inc_stats_size_limit_bytes</codeph> setting should be
+ increased with care. A big value for the setting, such as 1 GB or
+ more, can result in a spike in heap usage as well as a crash of
+ Impala.
+ </note>
+ </conbody>
+ </concept>
</concept>
[impala] 01/08: IMPALA-7174: fix test_cancellation for RELEASE
builds
Posted by ta...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tarmstrong pushed a commit to branch 2.x
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 31c3640f8858d30af9b4450b5ee72532b783ac45
Author: Tim Armstrong <ta...@cloudera.com>
AuthorDate: Thu Jun 14 14:21:06 2018 -0700
IMPALA-7174: fix test_cancellation for RELEASE builds
The test was DOA when run against a release build because the debug
actions that it depends on were disabled. The fix is to enable the
debug actions for release builds, similar to other debug actions.
I assume the original motivation of the NDEBUG checks was to avoid
adding overhead to release builds. The cost is minimised by quickly
checking whether the string is empty before proceeding with any
further work.
Also remove wonky exception handling - the test was swallowing
exceptions but we don't expect that code to throw exceptions.
Testing:
Looped the test on a release build.
Change-Id: I41da7b5ac58a468a8ed117777969906f63df6d4b
Reviewed-on: http://gerrit.cloudera.org:8080/10722
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Reviewed-on: http://gerrit.cloudera.org:8080/12504
Reviewed-by: Quanlong Huang <hu...@gmail.com>
Reviewed-by: Tim Armstrong <ta...@cloudera.com>
---
be/src/scheduling/admission-controller.cc | 4 ----
be/src/service/client-request-state.cc | 6 ------
be/src/util/debug-util.cc | 4 +---
be/src/util/debug-util.h | 15 +++++++++++----
tests/custom_cluster/test_admission_controller.py | 6 ++----
5 files changed, 14 insertions(+), 21 deletions(-)
diff --git a/be/src/scheduling/admission-controller.cc b/be/src/scheduling/admission-controller.cc
index f94d454..9a8d2ba 100644
--- a/be/src/scheduling/admission-controller.cc
+++ b/be/src/scheduling/admission-controller.cc
@@ -41,10 +41,8 @@ DEFINE_int64(queue_wait_timeout_ms, 60 * 1000, "Maximum amount of time (in "
namespace impala {
-#ifndef NDEBUG
// Available 'sleep_label' string that can be used in the debug_action query option.
static const string SLEEP_AFTER_ADMISSION_OUTCOME_MS = "SLEEP_AFTER_ADMISSION_OUTCOME_MS";
-#endif
/// Convenience method.
std::string PrintBytes(int64_t value) {
@@ -626,9 +624,7 @@ Status AdmissionController::AdmitQuery(QuerySchedule* schedule,
schedule->summary_profile()->AddInfoString(PROFILE_INFO_KEY_QUEUE_DETAIL,
Substitute(PROFILE_INFO_VAL_QUEUE_DETAIL, wait_time_ms, not_admitted_reason));
-#ifndef NDEBUG
SleepIfSetInDebugOptions(schedule->query_options(), SLEEP_AFTER_ADMISSION_OUTCOME_MS);
-#endif
{
lock_guard<mutex> lock(admission_ctrl_lock_);
diff --git a/be/src/service/client-request-state.cc b/be/src/service/client-request-state.cc
index 2e99e1d..09ad4e4 100644
--- a/be/src/service/client-request-state.cc
+++ b/be/src/service/client-request-state.cc
@@ -64,12 +64,10 @@ static const string TABLES_MISSING_STATS_KEY = "Tables Missing Stats";
static const string TABLES_WITH_CORRUPT_STATS_KEY = "Tables With Corrupt Table Stats";
static const string TABLES_WITH_MISSING_DISK_IDS_KEY = "Tables With Missing Disk Ids";
-#ifndef NDEBUG
// Available 'sleep_label' strings that can be used in the debug_action query option.
static const string SLEEP_BEFORE_ADMISSION_MS = "SLEEP_BEFORE_ADMISSION_MS";
static const string SLEEP_AFTER_COORDINATOR_STARTS_MS =
"SLEEP_AFTER_COORDINATOR_STARTS_MS";
-#endif
ClientRequestState::ClientRequestState(
const TQueryCtx& query_ctx, ExecEnv* exec_env, Frontend* frontend,
@@ -490,9 +488,7 @@ Status ClientRequestState::ExecAsyncQueryOrDmlRequest(
}
void ClientRequestState::FinishExecQueryOrDmlRequest() {
-#ifndef NDEBUG
SleepIfSetInDebugOptions(schedule_->query_options(), SLEEP_BEFORE_ADMISSION_MS);
-#endif
if (exec_env_->admission_controller() != nullptr) {
Status admit_status = ExecEnv::GetInstance()->admission_controller()->AdmitQuery(
schedule_.get(), &admit_outcome_);
@@ -504,9 +500,7 @@ void ClientRequestState::FinishExecQueryOrDmlRequest() {
coord_.reset(new Coordinator(*schedule_, query_events_));
Status exec_status = coord_->Exec();
-#ifndef NDEBUG
SleepIfSetInDebugOptions(schedule_->query_options(), SLEEP_AFTER_COORDINATOR_STARTS_MS);
-#endif
bool cancelled = false;
Status cancellation_status;
diff --git a/be/src/util/debug-util.cc b/be/src/util/debug-util.cc
index 3113703..eb55a7c 100644
--- a/be/src/util/debug-util.cc
+++ b/be/src/util/debug-util.cc
@@ -285,8 +285,7 @@ string GetBackendString() {
return Substitute("$0:$1", FLAGS_hostname, FLAGS_be_port);
}
-#ifndef NDEBUG
-void SleepIfSetInDebugOptions(
+void SleepIfSetInDebugOptionsImpl(
const TQueryOptions& query_options, const string& sleep_label) {
vector<string> components;
boost::split(components, query_options.debug_action, boost::is_any_of(":"));
@@ -294,6 +293,5 @@ void SleepIfSetInDebugOptions(
SleepForMs(atoi(components[1].c_str()));
}
}
-#endif
}
diff --git a/be/src/util/debug-util.h b/be/src/util/debug-util.h
index 9a01228..813f460 100644
--- a/be/src/util/debug-util.h
+++ b/be/src/util/debug-util.h
@@ -107,15 +107,22 @@ std::string GetStackTrace();
/// Returns the backend name in "host:port" form suitable for human consumption.
std::string GetBackendString();
-#ifndef NDEBUG
+/// Slow path implementing SleepIfSetInDebugOptions() for the case where 'query_options'
+/// is non-empty.
+void SleepIfSetInDebugOptionsImpl(
+ const TQueryOptions& query_options, const string& sleep_label);
+
/// If sleep time is specified in the debug_action query option in the format
/// <sleep_label>:<sleep_time_ms>, where <sleep_label> is a string and <sleep_time_ms>
/// is an integer, and 'sleep_label' matches the one specified in the query option, then
/// this methods extracts the corresponding <sleep_time_ms> and initiates a sleep for that
/// many milliseconds.
-void SleepIfSetInDebugOptions(
- const TQueryOptions& query_options, const string& sleep_label);
-#endif
+static inline void SleepIfSetInDebugOptions(
+ const TQueryOptions& query_options, const string& sleep_label) {
+ // Make sure this is very cheap if debug actions are not in use.
+ if (LIKELY(query_options.debug_action.empty())) return;
+ return SleepIfSetInDebugOptionsImpl(query_options, sleep_label);
+}
// FILE_CHECKs are conditions that we expect to be true but could fail due to a malformed
// input file. They differentiate these cases from DCHECKs, which indicate conditions that
diff --git a/tests/custom_cluster/test_admission_controller.py b/tests/custom_cluster/test_admission_controller.py
index 2beb60c..209991b 100644
--- a/tests/custom_cluster/test_admission_controller.py
+++ b/tests/custom_cluster/test_admission_controller.py
@@ -547,10 +547,6 @@ class TestAdmissionController(TestAdmissionControllerBase, HS2TestSuite):
client.close_query(handle)
queued_profile = client.get_runtime_profile(queued_query_handle)
assert "Admission result: Cancelled (queued)" in queued_profile
- except Exception as e:
- print e.args
- finally:
- client.close()
for i in self.cluster.impalads:
i.service.wait_for_metric_value("impala-server.num-fragments-in-flight", 0)
assert self.cluster.impalads[0].service.get_metric_value(
@@ -559,6 +555,8 @@ class TestAdmissionController(TestAdmissionControllerBase, HS2TestSuite):
"admission-controller.total-admitted.default-pool") == 3
assert self.cluster.impalads[0].service.get_metric_value(
"admission-controller.total-queued.default-pool") == 2
+ finally:
+ client.close()
class TestAdmissionControllerStress(TestAdmissionControllerBase):
"""Submits a number of queries (parameterized) with some delay between submissions
[impala] 02/08: IMPALA-7135. Skeleton implementation of LocalCatalog
Posted by ta...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tarmstrong pushed a commit to branch 2.x
in repository https://gitbox.apache.org/repos/asf/impala.git
commit c561fcc9225312ac7b1b1df4fbab912ef8d8d905
Author: Todd Lipcon <to...@cloudera.com>
AuthorDate: Wed Jun 6 15:00:13 2018 -0700
IMPALA-7135. Skeleton implementation of LocalCatalog
This adds some of the high level classes for implementing the local
catalog:
- LocalCatalog is the top level implementation. The plan is to
instantiate this once per query, so that no thread safety is required.
- It loads metadata from a MetaProvider interface. The current
implementation fetches directly from HMS and provides no caching. A
future subtask will add a CachingMetaProvider implementation.
Separating out caching will make it easier to experiment with
different policies or storage mechanisms.
- It instantiates LocalDb and LocalTable objects to implement FeDb and
FeTable. These are mostly stubbed out except for the most basic
functionality. Functionality will be filled in incrementally in
further patches.
Since it's not yet possible to hook this up to most of the existing
tests, a very simple new unit test is included to cover the bits of
functionality that are not stubbed out. I didn't concentrate on too much
test coverage here, since once we've implemented more functionality we
can switch over all of the existing tests to get coverage of the new
implementation.
Change-Id: Iab653371188b21c72f50ee1ec4e94950aa6fb9ee
Reviewed-on: http://gerrit.cloudera.org:8080/10627
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
.../java/org/apache/impala/catalog/Catalog.java | 8 +-
.../org/apache/impala/catalog/CatalogObject.java | 3 +-
.../main/java/org/apache/impala/catalog/FeDb.java | 3 +-
.../catalog/{CatalogObject.java => HasName.java} | 26 +--
.../impala/catalog/local/DirectMetaProvider.java | 83 ++++++++
.../apache/impala/catalog/local/LocalCatalog.java | 212 +++++++++++++++++++++
.../catalog/local/LocalCatalogException.java | 34 ++++
.../org/apache/impala/catalog/local/LocalDb.java | 173 +++++++++++++++++
.../apache/impala/catalog/local/LocalTable.java | 160 ++++++++++++++++
.../apache/impala/catalog/local/MetaProvider.java | 48 +++++
.../impala/catalog/local/LocalCatalogTest.java | 76 ++++++++
11 files changed, 800 insertions(+), 26 deletions(-)
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 82b411a..7403bc2 100644
--- a/fe/src/main/java/org/apache/impala/catalog/Catalog.java
+++ b/fe/src/main/java/org/apache/impala/catalog/Catalog.java
@@ -332,7 +332,7 @@ public abstract class Catalog {
* The results are sorted in String.CASE_INSENSITIVE_ORDER.
* matcher must not be null.
*/
- private List<String> filterStringsByPattern(Iterable<String> candidates,
+ public static List<String> filterStringsByPattern(Iterable<String> candidates,
PatternMatcher matcher) {
Preconditions.checkNotNull(matcher);
List<String> filtered = Lists.newArrayList();
@@ -343,9 +343,9 @@ public abstract class Catalog {
return filtered;
}
- private static class CatalogObjectOrder implements Comparator<CatalogObject> {
+ private static class CatalogObjectOrder implements Comparator<HasName> {
@Override
- public int compare(CatalogObject o1, CatalogObject o2) {
+ public int compare(HasName o1, HasName o2) {
return String.CASE_INSENSITIVE_ORDER.compare(o1.getName(), o2.getName());
}
}
@@ -357,7 +357,7 @@ public abstract class Catalog {
* The results are sorted in CATALOG_OBJECT_ORDER.
* matcher must not be null.
*/
- private <T extends CatalogObject> List<T> filterCatalogObjectsByPattern(
+ public static <T extends HasName> List<T> filterCatalogObjectsByPattern(
Iterable<? extends T> candidates, PatternMatcher matcher) {
Preconditions.checkNotNull(matcher);
List<T> filtered = Lists.newArrayList();
diff --git a/fe/src/main/java/org/apache/impala/catalog/CatalogObject.java b/fe/src/main/java/org/apache/impala/catalog/CatalogObject.java
index cc4c495..34310c3 100644
--- a/fe/src/main/java/org/apache/impala/catalog/CatalogObject.java
+++ b/fe/src/main/java/org/apache/impala/catalog/CatalogObject.java
@@ -22,11 +22,12 @@ import org.apache.impala.thrift.TCatalogObjectType;
/**
* Interface that all catalog objects implement.
*/
-public interface CatalogObject {
+public interface CatalogObject extends HasName {
// Returns the TCatalogObject type of this Catalog object.
public TCatalogObjectType getCatalogObjectType();
// Returns the unqualified object name.
+ @Override
public String getName();
// Returns the unique name of this catalog object.
diff --git a/fe/src/main/java/org/apache/impala/catalog/FeDb.java b/fe/src/main/java/org/apache/impala/catalog/FeDb.java
index 8de877f..111fbd0 100644
--- a/fe/src/main/java/org/apache/impala/catalog/FeDb.java
+++ b/fe/src/main/java/org/apache/impala/catalog/FeDb.java
@@ -26,10 +26,11 @@ import org.apache.impala.util.PatternMatcher;
/**
* Frontend interface for interacting with a database.
*/
-public interface FeDb {
+public interface FeDb extends HasName {
/**
* @return the name of the database
*/
+ @Override
String getName();
/**
diff --git a/fe/src/main/java/org/apache/impala/catalog/CatalogObject.java b/fe/src/main/java/org/apache/impala/catalog/HasName.java
similarity index 56%
copy from fe/src/main/java/org/apache/impala/catalog/CatalogObject.java
copy to fe/src/main/java/org/apache/impala/catalog/HasName.java
index cc4c495..594d541 100644
--- a/fe/src/main/java/org/apache/impala/catalog/CatalogObject.java
+++ b/fe/src/main/java/org/apache/impala/catalog/HasName.java
@@ -17,27 +17,13 @@
package org.apache.impala.catalog;
-import org.apache.impala.thrift.TCatalogObjectType;
-
/**
- * Interface that all catalog objects implement.
+ * Interface for named catalog objects, used for sorting and pattern
+ * matching.
*/
-public interface CatalogObject {
- // Returns the TCatalogObject type of this Catalog object.
- public TCatalogObjectType getCatalogObjectType();
-
- // Returns the unqualified object name.
+public interface HasName {
+ /**
+ * @return the unqualified name of the object
+ */
public String getName();
-
- // Returns the unique name of this catalog object.
- public String getUniqueName();
-
- // Returns the version of this catalog object.
- public long getCatalogVersion();
-
- // Sets the version of this catalog object.
- public void setCatalogVersion(long newVersion);
-
- // Returns true if this CatalogObject has had its metadata loaded, false otherwise.
- public boolean isLoaded();
}
diff --git a/fe/src/main/java/org/apache/impala/catalog/local/DirectMetaProvider.java b/fe/src/main/java/org/apache/impala/catalog/local/DirectMetaProvider.java
new file mode 100644
index 0000000..0867091
--- /dev/null
+++ b/fe/src/main/java/org/apache/impala/catalog/local/DirectMetaProvider.java
@@ -0,0 +1,83 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.impala.catalog.local;
+
+import org.apache.hadoop.hive.metastore.api.Database;
+import org.apache.hadoop.hive.metastore.api.MetaException;
+import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.hive.metastore.api.UnknownDBException;
+import org.apache.impala.catalog.MetaStoreClientPool;
+import org.apache.impala.catalog.MetaStoreClientPool.MetaStoreClient;
+import org.apache.impala.service.BackendConfig;
+import org.apache.impala.thrift.TBackendGflags;
+import org.apache.thrift.TException;
+
+import com.google.common.collect.ImmutableList;
+
+/**
+ * Metadata provider which calls out directly to the source systems
+ * (filesystem, HMS, etc) with no caching.
+ */
+class DirectMetaProvider implements MetaProvider {
+ private static MetaStoreClientPool msClientPool_;
+
+ DirectMetaProvider() {
+ initMsClientPool();
+ }
+
+ private static synchronized void initMsClientPool() {
+ // Lazy-init the metastore client pool based on the backend configuration.
+ // TODO(todd): this should probably be a process-wide singleton.
+ if (msClientPool_ == null) {
+ TBackendGflags cfg = BackendConfig.INSTANCE.getBackendCfg();
+ msClientPool_ = new MetaStoreClientPool(cfg.num_metadata_loading_threads,
+ cfg.initial_hms_cnxn_timeout_s);
+ }
+ }
+
+ @Override
+ public ImmutableList<String> loadDbList() throws TException {
+ try (MetaStoreClient c = msClientPool_.getClient()) {
+ return ImmutableList.copyOf(c.getHiveClient().getAllDatabases());
+ }
+ }
+
+ @Override
+ public Database loadDb(String dbName) throws TException {
+ try (MetaStoreClient c = msClientPool_.getClient()) {
+ return c.getHiveClient().getDatabase(dbName);
+ }
+ }
+
+ @Override
+ public ImmutableList<String> loadTableNames(String dbName)
+ throws MetaException, UnknownDBException, TException {
+ try (MetaStoreClient c = msClientPool_.getClient()) {
+ return ImmutableList.copyOf(c.getHiveClient().getAllTables(dbName));
+ }
+ }
+
+ @Override
+ public Table loadTable(String dbName, String tableName)
+ throws MetaException, NoSuchObjectException, TException {
+ try (MetaStoreClient c = msClientPool_.getClient()) {
+ return c.getHiveClient().getTable(dbName, tableName);
+ }
+ }
+}
diff --git a/fe/src/main/java/org/apache/impala/catalog/local/LocalCatalog.java b/fe/src/main/java/org/apache/impala/catalog/local/LocalCatalog.java
new file mode 100644
index 0000000..beac563
--- /dev/null
+++ b/fe/src/main/java/org/apache/impala/catalog/local/LocalCatalog.java
@@ -0,0 +1,212 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.impala.catalog.local;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.impala.analysis.TableName;
+import org.apache.impala.catalog.AuthorizationPolicy;
+import org.apache.impala.catalog.BuiltinsDb;
+import org.apache.impala.catalog.Catalog;
+import org.apache.impala.catalog.CatalogException;
+import org.apache.impala.catalog.DatabaseNotFoundException;
+import org.apache.impala.catalog.Db;
+import org.apache.impala.catalog.FeCatalog;
+import org.apache.impala.catalog.FeDataSource;
+import org.apache.impala.catalog.FeDb;
+import org.apache.impala.catalog.FeFsPartition;
+import org.apache.impala.catalog.FeTable;
+import org.apache.impala.catalog.Function;
+import org.apache.impala.catalog.Function.CompareMode;
+import org.apache.impala.catalog.HdfsCachePool;
+import org.apache.impala.catalog.ImpaladCatalog;
+import org.apache.impala.thrift.TCatalogObject;
+import org.apache.impala.thrift.TPartitionKeyValue;
+import org.apache.impala.thrift.TUniqueId;
+import org.apache.impala.util.PatternMatcher;
+import org.apache.thrift.TException;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Maps;
+
+/**
+ * Implementation of FeCatalog which runs within the impalad and fetches metadata
+ * on-demand using a {@link MetaProvider} instance.
+ *
+ * This class should be instantiated once per query. As the catalog is queried,
+ * it lazy-loads those pieces of metadata into the instance and instantiates
+ * appropriate catalog object implementations. This provides caching within a
+ * query -- multiple calls to fetch a given database, table, etc, will return
+ * the same instance. However, this class does not inherently provide any caching
+ * outside the scope of a single query: that caching should be performed at the
+ * level of the MetaProvider.
+ *
+ * This class is not thread-safe, nor are any of the catalog object implementations
+ * returned from its methods.
+ */
+public class LocalCatalog implements FeCatalog {
+ private final MetaProvider metaProvider_;
+ private Map<String, FeDb> dbs_ = Maps.newHashMap();
+ private static final Db builtinsDb_ = new BuiltinsDb(ImpaladCatalog.BUILTINS_DB);
+
+ public LocalCatalog(MetaProvider metaProvider) {
+ metaProvider_ = Preconditions.checkNotNull(metaProvider);
+ }
+
+ @Override
+ public List<? extends FeDb> getDbs(PatternMatcher matcher) {
+ loadDbs();
+ return Catalog.filterCatalogObjectsByPattern(dbs_.values(), matcher);
+ }
+
+ private void loadDbs() {
+ if (!dbs_.isEmpty()) return;
+ Map<String, FeDb> dbs = Maps.newHashMap();
+ List<String> names;
+ try {
+ names = metaProvider_.loadDbList();
+ } catch (TException e) {
+ throw new LocalCatalogException("Unable to load database names", e);
+ }
+ for (String dbName : names) {
+ dbName = dbName.toLowerCase();
+ if (dbs_.containsKey(dbName)) {
+ dbs.put(dbName, dbs_.get(dbName));
+ } else {
+ dbs.put(dbName, new LocalDb(this, dbName));
+ }
+ }
+ dbs.put(builtinsDb_.getName(), builtinsDb_);
+ dbs_ = dbs;
+ }
+
+
+ @Override
+ public List<String> getTableNames(String dbName, PatternMatcher matcher)
+ throws DatabaseNotFoundException {
+ return Catalog.filterStringsByPattern(getDbOrThrow(dbName).getAllTableNames(), matcher);
+ }
+
+ @Override
+ public FeTable getTable(String dbName, String tableName)
+ throws DatabaseNotFoundException {
+ return getDbOrThrow(dbName).getTable(tableName);
+ }
+
+ @Override
+ public TCatalogObject getTCatalogObject(TCatalogObject objectDesc)
+ throws CatalogException {
+ // TODO(todd): this probably makes the /catalog page not load with an error.
+ // We should probably disable that page in local-catalog mode.
+ throw new UnsupportedOperationException("LocalCatalog.getTCatalogObject");
+ }
+
+ @Override
+ public FeDb getDb(String db) {
+ loadDbs();
+ return dbs_.get(db);
+ }
+
+ private FeDb getDbOrThrow(String dbName) throws DatabaseNotFoundException {
+ Preconditions.checkNotNull(dbName);
+ FeDb db = getDb(dbName);
+ if (db == null) {
+ throw new DatabaseNotFoundException("Database '" + dbName + "' not found");
+ }
+ return db;
+ }
+
+ @Override
+ public FeFsPartition getHdfsPartition(
+ String db, String tbl, List<TPartitionKeyValue> partition_spec)
+ throws CatalogException {
+ throw new UnsupportedOperationException("TODO");
+ }
+
+ @Override
+ public List<? extends FeDataSource> getDataSources(
+ PatternMatcher createHivePatternMatcher) {
+ throw new UnsupportedOperationException("TODO");
+ }
+
+ @Override
+ public FeDataSource getDataSource(String dataSourceName) {
+ throw new UnsupportedOperationException("TODO");
+ }
+
+ @Override
+ public Function getFunction(Function desc, CompareMode mode) {
+ FeDb db = getDb(desc.dbName());
+ if (db == null) return null;
+ return db.getFunction(desc, mode);
+ }
+
+ @Override
+ public HdfsCachePool getHdfsCachePool(String poolName) {
+ throw new UnsupportedOperationException("TODO");
+ }
+
+ @Override
+ public void prioritizeLoad(Set<TableName> tableNames) {
+ // No-op for local catalog.
+ }
+
+ @Override
+ public void waitForCatalogUpdate(long timeoutMs) {
+ // No-op for local catalog.
+ }
+
+ @Override
+ public Path getTablePath(Table msTbl) {
+ throw new UnsupportedOperationException("TODO");
+ }
+
+ @Override
+ public TUniqueId getCatalogServiceId() {
+ throw new UnsupportedOperationException("TODO");
+ }
+
+ @Override
+ public AuthorizationPolicy getAuthPolicy() {
+ throw new UnsupportedOperationException("TODO");
+ }
+
+ @Override
+ public String getDefaultKuduMasterHosts() {
+ throw new UnsupportedOperationException("TODO");
+ }
+
+ @Override
+ public boolean isReady() {
+ // We are always ready.
+ return true;
+ }
+
+ @Override
+ public void setIsReady(boolean isReady) {
+ // No-op for local catalog.
+ }
+
+ MetaProvider getMetaProvider() {
+ return metaProvider_;
+ }
+}
\ No newline at end of file
diff --git a/fe/src/main/java/org/apache/impala/catalog/local/LocalCatalogException.java b/fe/src/main/java/org/apache/impala/catalog/local/LocalCatalogException.java
new file mode 100644
index 0000000..59e0be4
--- /dev/null
+++ b/fe/src/main/java/org/apache/impala/catalog/local/LocalCatalogException.java
@@ -0,0 +1,34 @@
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.impala.catalog.local;
+
+
+/**
+ * Exception indicating an error loading catalog information into
+ * the impalad.
+ *
+ * TODO(todd): should this be changed to a checked exception? The frontend
+ * interfaces don't have appropriate "throws" clauses to do so without some
+ * significant virality.
+ */
+public class LocalCatalogException extends RuntimeException {
+ private static final long serialVersionUID = -429271377417249918L;
+
+ public LocalCatalogException(String msg, Throwable cause) {
+ super(msg, cause);
+ }
+}
diff --git a/fe/src/main/java/org/apache/impala/catalog/local/LocalDb.java b/fe/src/main/java/org/apache/impala/catalog/local/LocalDb.java
new file mode 100644
index 0000000..1bbe9b8
--- /dev/null
+++ b/fe/src/main/java/org/apache/impala/catalog/local/LocalDb.java
@@ -0,0 +1,173 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.impala.catalog.local;
+
+import java.util.List;
+import java.util.Map;
+
+import org.apache.hadoop.hive.metastore.api.Database;
+import org.apache.impala.catalog.FeDb;
+import org.apache.impala.catalog.FeTable;
+import org.apache.impala.catalog.Function;
+import org.apache.impala.catalog.Function.CompareMode;
+import org.apache.impala.thrift.TDatabase;
+import org.apache.impala.thrift.TFunctionCategory;
+import org.apache.impala.util.PatternMatcher;
+import org.apache.thrift.TException;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Maps;
+
+/**
+ * Database instance loaded from {@link LocalCatalog}.
+ *
+ * This class is not thread-safe. A new instance is created for
+ * each catalog instance.
+ */
+class LocalDb implements FeDb {
+ private final LocalCatalog catalog_;
+ /** The lower-case name of the database. */
+ private final String name_;
+ private Database msDb_;
+
+ /**
+ * Map from lower-cased table name to table object. Values will be
+ * null for tables which have not yet been loaded.
+ */
+ private Map<String, LocalTable> tables_;
+
+ public LocalDb(LocalCatalog catalog, String dbName) {
+ Preconditions.checkNotNull(catalog);
+ Preconditions.checkNotNull(dbName);
+ Preconditions.checkArgument(dbName.toLowerCase().equals(dbName));
+ this.catalog_ = catalog;
+ this.name_ = dbName;
+ }
+
+ @Override
+ public String getName() {
+ return name_;
+ }
+
+ @Override
+ public Database getMetaStoreDb() {
+ if (msDb_ == null) {
+ try {
+ msDb_ = catalog_.getMetaProvider().loadDb(name_);
+ } catch (TException e) {
+ throw new LocalCatalogException(String.format(
+ "Could not load database '%s' from HMS", name_), e);
+ }
+ }
+ return msDb_;
+ }
+
+ @Override
+ public boolean containsTable(String tableName) {
+ loadTableNames();
+ return tables_.containsKey(tableName.toLowerCase());
+ }
+
+ @Override
+ public FeTable getTable(String tblName) {
+ Preconditions.checkNotNull(tblName);
+ Preconditions.checkArgument(tblName.toLowerCase().equals(tblName));
+ loadTableNames();
+ if (!tables_.containsKey(tblName)) {
+ // Table doesn't exist.
+ return null;
+ }
+ LocalTable tbl = tables_.get(tblName);
+ if (tbl == null) {
+ // The table exists but hasn't been loaded yet.
+ tbl = LocalTable.load(this, tblName);
+ tables_.put(tblName, tbl);
+ }
+ return tbl;
+ }
+
+ @Override
+ public List<String> getAllTableNames() {
+ loadTableNames();
+ return ImmutableList.copyOf(tables_.keySet());
+ }
+
+ /**
+ * Populate the 'tables_' map if it is not already populated.
+ *
+ * The map is populated with appropriate keys but null values which
+ * will be replaced on-demand.
+ */
+ private void loadTableNames() {
+ if (tables_ != null) return;
+ Map<String, LocalTable> newMap = Maps.newHashMap();
+ try {
+ List<String> names = catalog_.getMetaProvider().loadTableNames(name_);
+ for (String tableName : names) {
+ newMap.put(tableName.toLowerCase(), null);
+ }
+ } catch (TException e) {
+ throw new LocalCatalogException(String.format(
+ "Could not load table names for database '%s' from HMS", name_), e);
+ }
+ tables_ = newMap;
+ }
+
+ @Override
+ public boolean isSystemDb() {
+ return false;
+ }
+
+ @Override
+ public Function getFunction(Function desc, CompareMode mode) {
+ throw new UnsupportedOperationException("TODO");
+ }
+
+ @Override
+ public List<Function> getFunctions(String functionName) {
+ throw new UnsupportedOperationException("TODO");
+ }
+
+ @Override
+ public List<Function> getFunctions(
+ TFunctionCategory category, String function) {
+ throw new UnsupportedOperationException("TODO");
+ }
+
+ @Override
+ public List<Function> getFunctions(
+ TFunctionCategory category, PatternMatcher patternMatcher) {
+ throw new UnsupportedOperationException("TODO");
+ }
+
+ @Override
+ public int numFunctions() {
+ throw new UnsupportedOperationException("TODO");
+ }
+
+ @Override
+ public boolean containsFunction(String function) {
+ throw new UnsupportedOperationException("TODO");
+ }
+
+ @Override
+ public TDatabase toThrift() {
+ throw new UnsupportedOperationException("TODO");
+ }
+}
diff --git a/fe/src/main/java/org/apache/impala/catalog/local/LocalTable.java b/fe/src/main/java/org/apache/impala/catalog/local/LocalTable.java
new file mode 100644
index 0000000..58c0580
--- /dev/null
+++ b/fe/src/main/java/org/apache/impala/catalog/local/LocalTable.java
@@ -0,0 +1,160 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.impala.catalog.local;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.impala.analysis.TableName;
+import org.apache.impala.catalog.ArrayType;
+import org.apache.impala.catalog.Column;
+import org.apache.impala.catalog.FeDb;
+import org.apache.impala.catalog.FeTable;
+import org.apache.impala.thrift.TCatalogObjectType;
+import org.apache.impala.thrift.TTableDescriptor;
+import org.apache.impala.thrift.TTableStats;
+
+import com.google.common.base.Preconditions;
+
+/**
+ * Table instance loaded from {@link LocalCatalog}.
+ *
+ * This class is not thread-safe. A new instance is created for
+ * each catalog instance.
+ */
+class LocalTable implements FeTable {
+ private final LocalDb db_;
+ /** The lower-case name of the table. */
+ private final String name_;
+
+ public static LocalTable load(LocalDb db, String tblName) {
+ // TODO: change this function to instantiate the appropriate
+ // subclass based on the table type (eg view, hbase table, etc)
+ return new LocalTable(db, tblName);
+ }
+
+ public LocalTable(LocalDb db, String tblName) {
+ Preconditions.checkNotNull(db);
+ Preconditions.checkNotNull(tblName);
+ Preconditions.checkArgument(tblName.toLowerCase().equals(tblName));
+ this.db_ = db;
+ this.name_ = tblName;
+ }
+
+ @Override
+ public boolean isLoaded() {
+ return true;
+ }
+
+ @Override
+ public Table getMetaStoreTable() {
+ throw new UnsupportedOperationException("TODO");
+ }
+
+ @Override
+ public String getStorageHandlerClassName() {
+ throw new UnsupportedOperationException("TODO");
+ }
+
+ @Override
+ public TCatalogObjectType getCatalogObjectType() {
+ throw new UnsupportedOperationException("TODO");
+ }
+
+ @Override
+ public String getName() {
+ return name_;
+ }
+
+ @Override
+ public String getFullName() {
+ return db_.getName() + "." + name_;
+ }
+
+ @Override
+ public TableName getTableName() {
+ return new TableName(db_.getName(), name_);
+ }
+
+ @Override
+ public ArrayList<Column> getColumns() {
+ throw new UnsupportedOperationException("TODO");
+ }
+
+ @Override
+ public List<Column> getColumnsInHiveOrder() {
+ throw new UnsupportedOperationException("TODO");
+ }
+
+ @Override
+ public List<String> getColumnNames() {
+ throw new UnsupportedOperationException("TODO");
+ }
+
+ @Override
+ public List<Column> getClusteringColumns() {
+ throw new UnsupportedOperationException("TODO");
+ }
+
+ @Override
+ public List<Column> getNonClusteringColumns() {
+ throw new UnsupportedOperationException("TODO");
+ }
+
+ @Override
+ public int getNumClusteringCols() {
+ throw new UnsupportedOperationException("TODO");
+ }
+
+ @Override
+ public boolean isClusteringColumn(Column c) {
+ throw new UnsupportedOperationException("TODO");
+ }
+
+ @Override
+ public Column getColumn(String name) {
+ throw new UnsupportedOperationException("TODO");
+ }
+
+ @Override
+ public ArrayType getType() {
+ throw new UnsupportedOperationException("TODO");
+ }
+
+ @Override
+ public FeDb getDb() {
+ return db_;
+ }
+
+ @Override
+ public long getNumRows() {
+ throw new UnsupportedOperationException("TODO");
+ }
+
+ @Override
+ public TTableStats getTTableStats() {
+ throw new UnsupportedOperationException("TODO");
+ }
+
+ @Override
+ public TTableDescriptor toThriftDescriptor(int tableId, Set<Long> referencedPartitions) {
+ throw new UnsupportedOperationException("TODO");
+ }
+}
diff --git a/fe/src/main/java/org/apache/impala/catalog/local/MetaProvider.java b/fe/src/main/java/org/apache/impala/catalog/local/MetaProvider.java
new file mode 100644
index 0000000..cd8d3b9
--- /dev/null
+++ b/fe/src/main/java/org/apache/impala/catalog/local/MetaProvider.java
@@ -0,0 +1,48 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.impala.catalog.local;
+
+import org.apache.hadoop.hive.metastore.api.Database;
+import org.apache.hadoop.hive.metastore.api.MetaException;
+import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.hive.metastore.api.UnknownDBException;
+import org.apache.thrift.TException;
+
+import com.google.common.collect.ImmutableList;
+
+/**
+ * Interface for loading metadata. See {@link LocalCatalog} for an example.
+ *
+ * Implementations may directly access the metadata from the source systems
+ * or may include caching, etc.
+ *
+ * TODO(IMPALA-7127): expand this to include file metadata, sentry metadata,
+ * etc.
+ */
+interface MetaProvider {
+ ImmutableList<String> loadDbList() throws TException;
+
+ Database loadDb(String dbName) throws TException;
+
+ ImmutableList<String> loadTableNames(String dbName)
+ throws MetaException, UnknownDBException, TException;
+
+ Table loadTable(String dbName, String tableName)
+ throws NoSuchObjectException, MetaException, TException;
+}
diff --git a/fe/src/test/java/org/apache/impala/catalog/local/LocalCatalogTest.java b/fe/src/test/java/org/apache/impala/catalog/local/LocalCatalogTest.java
new file mode 100644
index 0000000..52ff255
--- /dev/null
+++ b/fe/src/test/java/org/apache/impala/catalog/local/LocalCatalogTest.java
@@ -0,0 +1,76 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.impala.catalog.local;
+
+import static org.junit.Assert.*;
+
+import java.util.Set;
+
+import org.apache.impala.catalog.FeDb;
+import org.apache.impala.catalog.FeTable;
+import org.apache.impala.util.PatternMatcher;
+import org.junit.Before;
+import org.junit.Test;
+
+import com.google.common.collect.ImmutableSet;
+
+public class LocalCatalogTest {
+
+ private LocalCatalog catalog_;
+
+ @Before
+ public void setupCatalog() {
+ catalog_ = new LocalCatalog(new DirectMetaProvider());
+ }
+
+ @Test
+ public void testDbs() throws Exception {
+ FeDb functionalDb = catalog_.getDb("functional");
+ assertNotNull(functionalDb);
+ FeDb functionalSeqDb = catalog_.getDb("functional_seq");
+ assertNotNull(functionalSeqDb);
+
+ Set<FeDb> dbs = ImmutableSet.copyOf(
+ catalog_.getDbs(PatternMatcher.MATCHER_MATCH_ALL));
+ assertTrue(dbs.contains(functionalDb));
+ assertTrue(dbs.contains(functionalSeqDb));
+
+ dbs = ImmutableSet.copyOf(
+ catalog_.getDbs(PatternMatcher.createHivePatternMatcher("*_seq")));
+ assertFalse(dbs.contains(functionalDb));
+ assertTrue(dbs.contains(functionalSeqDb));
+ }
+
+ @Test
+ public void testTables() throws Exception {
+ Set<String> names = ImmutableSet.copyOf(catalog_.getTableNames(
+ "functional", PatternMatcher.MATCHER_MATCH_ALL));
+ assertTrue(names.contains("alltypes"));
+
+ FeDb db = catalog_.getDb("functional");
+ assertNotNull(db);
+ FeTable t = catalog_.getTable("functional", "alltypes");
+ assertNotNull(t);
+ assertSame(t, db.getTable("alltypes"));
+ assertSame(db, t.getDb());
+ assertEquals("alltypes", t.getName());
+ assertEquals("functional", t.getDb().getName());
+ assertEquals("functional.alltypes", t.getFullName());
+ }
+
+}
[impala] 05/08: IMPALA-7130: impala-shell -b / --kerberos_host_fqdn
flag overrides value passed in via -i / --impalad
Posted by ta...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tarmstrong pushed a commit to branch 2.x
in repository https://gitbox.apache.org/repos/asf/impala.git
commit a8d2f2fe2f31dd04bd0e3731ef7f452bd061e5f2
Author: Vincent Tran <vt...@cloudera.com>
AuthorDate: Fri Jun 1 18:49:40 2018 -0400
IMPALA-7130: impala-shell -b / --kerberos_host_fqdn flag overrides value passed in via -i / --impalad
After additional testing around IMPALA-2782, it was discovered
that impala-shell starts the session displaying the expected
hostname (as passed -i flag) on the prompt. This gives the
impression that the load balancer was bypassed, however the
actual TSSLSocket is still created with the hostname passed
in via the -b or --kerberos_host_fqdn flag.
This change ensures that the hostname used to create the
TSSLSocket will always be the one passed in via the -i flag
on impala-shell. This change is required by IMPALA-2782.
Testing:
Using netcat, we verified that the impala daemon host[:port]
value passed into the -i/--impalad option is indeed the one
impala-shell tries to connect to in both cases (with and
without -b)
Change-Id: Ibee05bd0dbe8c6ae108b890f0ae0f6900149773a
Reviewed-on: http://gerrit.cloudera.org:8080/10580
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
shell/impala_client.py | 23 ++++++++++++++---------
tests/shell/test_shell_commandline.py | 35 ++++++++++++++++++++++++++++++++++-
2 files changed, 48 insertions(+), 10 deletions(-)
diff --git a/shell/impala_client.py b/shell/impala_client.py
index 5fa50b0..69c7699 100755
--- a/shell/impala_client.py
+++ b/shell/impala_client.py
@@ -63,7 +63,8 @@ class ImpalaClient(object):
kerberos_service_name="impala", use_ssl=False, ca_cert=None, user=None,
ldap_password=None, use_ldap=False):
self.connected = False
- self.impalad = impalad
+ self.impalad_host = impalad[0].encode('ascii', 'ignore')
+ self.impalad_port = int(impalad[1])
self.kerberos_host_fqdn = kerberos_host_fqdn
self.imp_service = None
self.transport = None
@@ -278,28 +279,32 @@ class ImpalaClient(object):
# sasl does not accept unicode strings, explicitly encode the string into ascii.
# The kerberos_host_fqdn option exposes the SASL client's hostname attribute to
# the user. impala-shell checks to ensure this host matches the host in the kerberos
- # principal. So in the presence of a load balancer, the its hostname is expected by
+ # principal. So when a load balancer is configured to be used, its hostname is expected by
# impala-shell. Setting this option to the load balancer hostname allows impala-shell to
# connect directly to an impalad.
if self.kerberos_host_fqdn is not None:
- host, port = (self.kerberos_host_fqdn.split(':')[0].encode('ascii', 'ignore'),
- int(self.impalad[1]))
+ sasl_host = self.kerberos_host_fqdn.split(':')[0].encode('ascii', 'ignore')
else:
- host, port = self.impalad[0].encode('ascii', 'ignore'), int(self.impalad[1])
+ sasl_host = self.impalad_host
+
+ # Always use the hostname and port passed in to -i / --impalad as the host for the purpose of
+ # creating the actual socket.
+ sock_host = self.impalad_host
+ sock_port = self.impalad_port
if self.use_ssl:
if self.ca_cert is None:
# No CA cert means don't try to verify the certificate
- sock = TSSLSocketWithWildcardSAN(host, port, validate=False)
+ sock = TSSLSocketWithWildcardSAN(sock_host, sock_port, validate=False)
else:
- sock = TSSLSocketWithWildcardSAN(host, port, validate=True, ca_certs=self.ca_cert)
+ sock = TSSLSocketWithWildcardSAN(sock_host, sock_port, validate=True, ca_certs=self.ca_cert)
else:
- sock = TSocket(host, port)
+ sock = TSocket(sock_host, sock_port)
if not (self.use_ldap or self.use_kerberos):
return TBufferedTransport(sock)
# Initializes a sasl client
def sasl_factory():
sasl_client = sasl.Client()
- sasl_client.setAttr("host", host)
+ sasl_client.setAttr("host", sasl_host)
if self.use_ldap:
sasl_client.setAttr("username", self.user)
sasl_client.setAttr("password", self.ldap_password)
diff --git a/tests/shell/test_shell_commandline.py b/tests/shell/test_shell_commandline.py
index 5fec7fb..aa936ac 100644
--- a/tests/shell/test_shell_commandline.py
+++ b/tests/shell/test_shell_commandline.py
@@ -21,8 +21,9 @@ import os
import pytest
import re
import signal
+import shlex
-from subprocess import call
+from subprocess import call, Popen, PIPE
from tests.common.impala_service import ImpaladService
from tests.common.impala_test_suite import ImpalaTestSuite
from tests.common.skip import SkipIf
@@ -609,3 +610,35 @@ class TestImpalaShell(ImpalaTestSuite):
def test_missing_query_file(self):
result = run_impala_shell_cmd('-f nonexistent.sql', expect_success=False)
assert "Could not open file 'nonexistent.sql'" in result.stderr
+
+ def test_socket_opening(self):
+ ''' Tests that the impala daemon will always open a socket against
+ the host[:port] specified by the -i option with or without the
+ -b option '''
+
+ impala_daemon_port = 42000
+ load_balancer_fqdn = "my-load-balancer.local"
+ ncat_timeout = 1
+ # Building an one-off shell cmd instead of using Util::ImpalaShell since we need
+ # to customize the impala daemon socket
+ shell_cmd = "%s/bin/impala-shell.sh" % (os.environ['IMPALA_HOME'])
+ args1 = "-i localhost:%d" % (impala_daemon_port,)
+ args2 = "-b %s" % (load_balancer_fqdn,)
+
+ # Verify that impala-shell tries to create a socket again localhost:42000 as
+ # specified by -i option without the -b option
+ impalad_sock = Popen(shlex.split("nc -lp %d -w %d" % (impala_daemon_port, ncat_timeout,)),
+ stdout = PIPE, stderr = PIPE)
+ impala_shell = Popen(shlex.split("%s %s" % (shell_cmd, args1, )))
+ impalad_sock_stdout, impalad_sock_stderr = impalad_sock.communicate()
+ expected_output = "PingImpalaService"
+ assert expected_output in impalad_sock_stdout
+
+ # Verify that impala-shell tries to create a socket again localhost:42000 as
+ # specified by -i option with the -b option
+
+ impalad_sock = Popen(shlex.split("nc -lp %d -w %d" % (impala_daemon_port, ncat_timeout,)),
+ stdout = PIPE, stderr = PIPE)
+ impala_shell = Popen(shlex.split("%s %s %s" % (shell_cmd, args1, args2, )))
+ impalad_sock_stdout, impalad_sock_stderr = impalad_sock.communicate()
+ assert expected_output in impalad_sock_stdout
[impala] 08/08: IMPALA-7137. Support configuring Frontend to use
LocalCatalog
Posted by ta...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tarmstrong pushed a commit to branch 2.x
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 800ac7d9f4dbceeb87bbc9e4a464a29fa34f99a4
Author: Todd Lipcon <to...@cloudera.com>
AuthorDate: Wed Jun 6 16:17:49 2018 -0700
IMPALA-7137. Support configuring Frontend to use LocalCatalog
This adds a new flag -use_local_catalog which is passed through to the
frontend and causes it to use LocalCatalog instead of ImpaladCatalog.
Additionally, when this flag is configured, the impalad does not
subscribe to catalog topic updates from the statestore.
The patch is slightly more complex than simply picking which class to
instantiate, because the lifecycle is designed a bit differently between
the two implementations:
- LocalCatalog is instantiated once per query/request.
- ImpaladCatalog is instantiated once and stateful across queries,
except when a full catalog update is received. This maintains the
current behavior for this implementation.
In order to abstract this difference in lifecycle from the frontend, I
introduced a new FeCatalogManager class with different implementations
for the two lifecycles. I also had to add a simple test implementation
since some tests rely on directly injecting a Catalog implementation
into the Frontend.
This patch also includes a few small changes to the local catalog
implementation objects to enable the impalad to start and accept
connections. With this patch, I was able to manually test as follows:
I started just the statestore and the impalad in the new mode:
- ./bin/start-statestored.sh
- ./bin/start-impalad.sh --use_local_catalog
I connected with impala-shell as usual and was able to run the most
simple queries:
- SHOW DATABASES;
- USE functional;
- SHOW TABLES;
All other functionality results in error messages due to the various
TODOs in the current skeleton implementation.
Change-Id: I8c9665bd031d23608740b23eef301970af9aa764
Reviewed-on: http://gerrit.cloudera.org:8080/10629
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Reviewed-by: Vuk Ercegovac <ve...@cloudera.com>
---
be/src/runtime/exec-env.cc | 4 +
be/src/service/impala-server.cc | 3 +-
be/src/util/backend-gflag-util.cc | 2 +
common/thrift/BackendGflags.thrift | 2 +
.../apache/impala/catalog/local/LocalCatalog.java | 7 +-
.../org/apache/impala/catalog/local/LocalDb.java | 4 +-
.../impala/service/DescribeResultFactory.java | 12 +-
.../apache/impala/service/FeCatalogManager.java | 157 +++++++++++++++++++++
.../java/org/apache/impala/service/Frontend.java | 104 +++++++-------
.../org/apache/impala/service/JniFrontend.java | 2 +-
.../java/org/apache/impala/service/MetadataOp.java | 8 +-
11 files changed, 241 insertions(+), 64 deletions(-)
diff --git a/be/src/runtime/exec-env.cc b/be/src/runtime/exec-env.cc
index 1a38bc7..1322bd8 100644
--- a/be/src/runtime/exec-env.cc
+++ b/be/src/runtime/exec-env.cc
@@ -84,6 +84,10 @@ DEFINE_bool(disable_admission_control, false, "Disables admission control.");
DEFINE_bool(use_krpc, true, "If true, use KRPC for the DataStream subsystem. "
"Otherwise use Thrift RPC.");
+DEFINE_bool_hidden(use_local_catalog, false,
+ "Use experimental implementation of a local catalog. If this is set, "
+ "the catalog service is not used and does not need to be started.");
+
DECLARE_int32(state_store_port);
DECLARE_int32(num_threads_per_core);
DECLARE_int32(num_cores);
diff --git a/be/src/service/impala-server.cc b/be/src/service/impala-server.cc
index 2026842..ce6c2c3 100644
--- a/be/src/service/impala-server.cc
+++ b/be/src/service/impala-server.cc
@@ -119,6 +119,7 @@ DECLARE_string(authorized_proxy_group_config_delimiter);
DECLARE_bool(abort_on_config_error);
DECLARE_bool(disk_spill_encryption);
DECLARE_bool(use_krpc);
+DECLARE_bool(use_local_catalog);
DEFINE_int32(beeswax_port, 21000, "port on which Beeswax client requests are served");
DEFINE_int32(hs2_port, 21050, "port on which HiveServer2 client requests are served");
@@ -364,7 +365,7 @@ ImpalaServer::ImpalaServer(ExecEnv* exec_env)
ABORT_IF_ERROR(
exec_env->subscriber()->AddTopic(Statestore::IMPALA_MEMBERSHIP_TOPIC, true, cb));
- if (FLAGS_is_coordinator) {
+ if (FLAGS_is_coordinator && !FLAGS_use_local_catalog) {
auto catalog_cb = [this] (const StatestoreSubscriber::TopicDeltaMap& state,
vector<TTopicDelta>* topic_updates) {
this->CatalogUpdateCallback(state, topic_updates);
diff --git a/be/src/util/backend-gflag-util.cc b/be/src/util/backend-gflag-util.cc
index acbdc6f..d8d945a 100644
--- a/be/src/util/backend-gflag-util.cc
+++ b/be/src/util/backend-gflag-util.cc
@@ -28,6 +28,7 @@ DECLARE_bool(load_catalog_in_background);
DECLARE_bool(load_auth_to_local_rules);
DECLARE_bool(enable_stats_extrapolation);
DECLARE_bool(enable_orc_scanner);
+DECLARE_bool(use_local_catalog);
DECLARE_int32(non_impala_java_vlog);
DECLARE_int32(num_metadata_loading_threads);
DECLARE_int32(max_hdfs_partitions_parallel_load);
@@ -60,6 +61,7 @@ Status GetThriftBackendGflags(JNIEnv* jni_env, jbyteArray* cfg_bytes) {
cfg.__set_authorization_policy_file(FLAGS_authorization_policy_file);
cfg.__set_load_catalog_in_background(FLAGS_load_catalog_in_background);
cfg.__set_enable_orc_scanner(FLAGS_enable_orc_scanner);
+ cfg.__set_use_local_catalog(FLAGS_use_local_catalog);
cfg.__set_server_name(FLAGS_server_name);
cfg.__set_sentry_config(FLAGS_sentry_config);
cfg.__set_authorization_policy_provider_class(
diff --git a/common/thrift/BackendGflags.thrift b/common/thrift/BackendGflags.thrift
index f645e3c..c8c0dea 100644
--- a/common/thrift/BackendGflags.thrift
+++ b/common/thrift/BackendGflags.thrift
@@ -77,4 +77,6 @@ struct TBackendGflags {
25: required bool enable_orc_scanner
26: required string authorized_proxy_group_config
+
+ 27: required bool use_local_catalog
}
diff --git a/fe/src/main/java/org/apache/impala/catalog/local/LocalCatalog.java b/fe/src/main/java/org/apache/impala/catalog/local/LocalCatalog.java
index beac563..2b246e6 100644
--- a/fe/src/main/java/org/apache/impala/catalog/local/LocalCatalog.java
+++ b/fe/src/main/java/org/apache/impala/catalog/local/LocalCatalog.java
@@ -68,6 +68,11 @@ public class LocalCatalog implements FeCatalog {
private Map<String, FeDb> dbs_ = Maps.newHashMap();
private static final Db builtinsDb_ = new BuiltinsDb(ImpaladCatalog.BUILTINS_DB);
+ public static FeCatalog create(String defaultKuduMasterHosts) {
+ // TODO(todd): store the kudu master hosts
+ return new LocalCatalog(new DirectMetaProvider());
+ }
+
public LocalCatalog(MetaProvider metaProvider) {
metaProvider_ = Preconditions.checkNotNull(metaProvider);
}
@@ -187,7 +192,7 @@ public class LocalCatalog implements FeCatalog {
@Override
public AuthorizationPolicy getAuthPolicy() {
- throw new UnsupportedOperationException("TODO");
+ return null; // TODO(todd): implement auth policy
}
@Override
diff --git a/fe/src/main/java/org/apache/impala/catalog/local/LocalDb.java b/fe/src/main/java/org/apache/impala/catalog/local/LocalDb.java
index 1bbe9b8..b3bf010 100644
--- a/fe/src/main/java/org/apache/impala/catalog/local/LocalDb.java
+++ b/fe/src/main/java/org/apache/impala/catalog/local/LocalDb.java
@@ -168,6 +168,8 @@ class LocalDb implements FeDb {
@Override
public TDatabase toThrift() {
- throw new UnsupportedOperationException("TODO");
+ TDatabase tdb = new TDatabase(name_);
+ tdb.setMetastore_db(getMetaStoreDb());
+ return tdb;
}
}
diff --git a/fe/src/main/java/org/apache/impala/service/DescribeResultFactory.java b/fe/src/main/java/org/apache/impala/service/DescribeResultFactory.java
index 9a5adf4..d1081c6 100644
--- a/fe/src/main/java/org/apache/impala/service/DescribeResultFactory.java
+++ b/fe/src/main/java/org/apache/impala/service/DescribeResultFactory.java
@@ -27,11 +27,11 @@ import org.apache.hadoop.hive.metastore.api.PrincipalType;
import org.apache.hadoop.hive.metastore.api.PrivilegeGrantInfo;
import org.apache.hadoop.hive.ql.metadata.formatting.MetaDataFormatUtils;
import org.apache.impala.catalog.Column;
-import org.apache.impala.catalog.Db;
+import org.apache.impala.catalog.FeDb;
+import org.apache.impala.catalog.FeTable;
import org.apache.impala.catalog.KuduColumn;
import org.apache.impala.catalog.StructField;
import org.apache.impala.catalog.StructType;
-import org.apache.impala.catalog.Table;
import org.apache.impala.thrift.TColumnValue;
import org.apache.impala.thrift.TDescribeOutputStyle;
import org.apache.impala.thrift.TDescribeResult;
@@ -51,7 +51,7 @@ public class DescribeResultFactory {
// Empty column used to format description output table.
private final static TColumnValue EMPTY = new TColumnValue().setString_val("");
- public static TDescribeResult buildDescribeDbResult(Db db,
+ public static TDescribeResult buildDescribeDbResult(FeDb db,
TDescribeOutputStyle outputFormat) {
switch (outputFormat) {
case MINIMAL: return describeDbMinimal(db);
@@ -67,7 +67,7 @@ public class DescribeResultFactory {
* Builds results for a DESCRIBE DATABASE <db> command. This consists of the database
* location and comment.
*/
- private static TDescribeResult describeDbMinimal(Db db) {
+ private static TDescribeResult describeDbMinimal(FeDb db) {
TDescribeResult descResult = new TDescribeResult();
org.apache.hadoop.hive.metastore.api.Database msDb = db.getMetaStoreDb();
@@ -122,7 +122,7 @@ public class DescribeResultFactory {
* Builds a TDescribeResult that contains the result of a DESCRIBE FORMATTED|EXTENDED
* DATABASE <db> command. Output all the database's properties.
*/
- private static TDescribeResult describeDbExtended(Db db) {
+ private static TDescribeResult describeDbExtended(FeDb db) {
TDescribeResult descResult = describeDbMinimal(db);
org.apache.hadoop.hive.metastore.api.Database msDb = db.getMetaStoreDb();
String ownerName = null;
@@ -184,7 +184,7 @@ public class DescribeResultFactory {
* Hive's MetadataFormatUtils class is used to build the results. filteredColumns is a
* list of columns the user is authorized to view.
*/
- public static TDescribeResult buildDescribeFormattedResult(Table table,
+ public static TDescribeResult buildDescribeFormattedResult(FeTable table,
List<Column> filteredColumns) {
TDescribeResult result = new TDescribeResult();
result.results = Lists.newArrayList();
diff --git a/fe/src/main/java/org/apache/impala/service/FeCatalogManager.java b/fe/src/main/java/org/apache/impala/service/FeCatalogManager.java
new file mode 100644
index 0000000..2bd44a4
--- /dev/null
+++ b/fe/src/main/java/org/apache/impala/service/FeCatalogManager.java
@@ -0,0 +1,157 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+package org.apache.impala.service;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.impala.catalog.CatalogException;
+import org.apache.impala.catalog.FeCatalog;
+import org.apache.impala.catalog.ImpaladCatalog;
+import org.apache.impala.catalog.local.LocalCatalog;
+import org.apache.impala.thrift.TUpdateCatalogCacheRequest;
+import org.apache.impala.thrift.TUpdateCatalogCacheResponse;
+import org.apache.thrift.TException;
+
+/**
+ * Manages the Catalog implementation used by the frontend.
+ *
+ * This class abstracts away the different lifecycles used by the LocalCatalog
+ * and the ImpaladCatalog. The former creates a new instance for each request or
+ * query, whereas the latter only creates a new instance upon receiving a full update
+ * from the catalogd via the statestore.
+ */
+public abstract class FeCatalogManager {
+ private static String DEFAULT_KUDU_MASTER_HOSTS =
+ BackendConfig.INSTANCE.getBackendCfg().kudu_master_hosts;
+
+ /**
+ * @return the appropriate implementation based on the current backend
+ * configuration.
+ */
+ public static FeCatalogManager createFromBackendConfig() {
+ if (BackendConfig.INSTANCE.getBackendCfg().use_local_catalog) {
+ return new LocalImpl();
+ } else {
+ return new CatalogdImpl();
+ }
+ }
+
+ /**
+ * Create a manager which always returns the same instance and does not permit
+ * updates from the statestore.
+ */
+ public static FeCatalogManager createForTests(FeCatalog testCatalog) {
+ return new TestImpl(testCatalog);
+ }
+
+ /**
+ * @return a Catalog instance to be used for a request or query. Depending
+ * on the catalog implementation this may either be a reused instance or a
+ * fresh one for each query.
+ */
+ abstract FeCatalog getOrCreateCatalog();
+
+ /**
+ * Update the Catalog based on an update from the state store. Only supported
+ * by the catalogd-based implementation.
+ */
+ abstract TUpdateCatalogCacheResponse updateCatalogCache(
+ TUpdateCatalogCacheRequest req) throws CatalogException, TException;
+
+ /**
+ * Implementation which creates ImpaladCatalog instances and expects to receive
+ * updates via the statestore. New instances are created only when full updates
+ * are received.
+ */
+ private static class CatalogdImpl extends FeCatalogManager {
+ private final AtomicReference<ImpaladCatalog> catalog_ =
+ new AtomicReference<>();
+
+ private CatalogdImpl() {
+ catalog_.set(createNewCatalog());
+ }
+
+ @Override
+ FeCatalog getOrCreateCatalog() {
+ return catalog_.get();
+ }
+
+ @Override
+ TUpdateCatalogCacheResponse updateCatalogCache(TUpdateCatalogCacheRequest req)
+ throws CatalogException, TException {
+ ImpaladCatalog catalog = catalog_.get();
+ if (req.is_delta) return catalog.updateCatalog(req);
+
+ // If this is not a delta, this update should replace the current
+ // Catalog contents so create a new catalog and populate it.
+ catalog = createNewCatalog();
+
+ TUpdateCatalogCacheResponse response = catalog.updateCatalog(req);
+
+ // Now that the catalog has been updated, replace the reference to
+ // catalog_. This ensures that clients don't see the catalog
+ // disappear. The catalog is guaranteed to be ready since updateCatalog() has a
+ // postcondition of isReady() == true.
+ catalog_.set(catalog);
+ return response;
+ }
+
+ private ImpaladCatalog createNewCatalog() {
+ return new ImpaladCatalog(DEFAULT_KUDU_MASTER_HOSTS);
+ }
+ }
+
+ /**
+ * Implementation which creates LocalCatalog instances. A new instance is
+ * created for each request or query.
+ */
+ private static class LocalImpl extends FeCatalogManager {
+ @Override
+ FeCatalog getOrCreateCatalog() {
+ return LocalCatalog.create(DEFAULT_KUDU_MASTER_HOSTS);
+ }
+
+ @Override
+ TUpdateCatalogCacheResponse updateCatalogCache(TUpdateCatalogCacheRequest req) {
+ throw new IllegalStateException(
+ "Unexpected call to updateCatalogCache() with local catalog enabled");
+ }
+ }
+
+ /**
+ * Implementation which returns a provided catalog instance, used by tests.
+ * No updates from the statestore are permitted.
+ */
+ private static class TestImpl extends FeCatalogManager {
+ private final FeCatalog catalog_;
+
+ TestImpl(FeCatalog catalog) {
+ catalog_ = catalog;
+ }
+
+ @Override
+ FeCatalog getOrCreateCatalog() {
+ return catalog_;
+ }
+
+ @Override
+ TUpdateCatalogCacheResponse updateCatalogCache(TUpdateCatalogCacheRequest req) {
+ throw new IllegalStateException(
+ "Unexpected call to updateCatalogCache() with a test catalog instance");
+ }
+ }
+}
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 ad51286..ab330fd 100644
--- a/fe/src/main/java/org/apache/impala/service/Frontend.java
+++ b/fe/src/main/java/org/apache/impala/service/Frontend.java
@@ -76,17 +76,16 @@ import org.apache.impala.catalog.Catalog;
import org.apache.impala.catalog.CatalogException;
import org.apache.impala.catalog.Column;
import org.apache.impala.catalog.DatabaseNotFoundException;
-import org.apache.impala.catalog.Db;
+import org.apache.impala.catalog.FeCatalog;
import org.apache.impala.catalog.FeDataSource;
import org.apache.impala.catalog.FeDataSourceTable;
import org.apache.impala.catalog.FeDb;
import org.apache.impala.catalog.FeFsTable;
+import org.apache.impala.catalog.FeTable;
import org.apache.impala.catalog.Function;
import org.apache.impala.catalog.HBaseTable;
-import org.apache.impala.catalog.HdfsTable;
import org.apache.impala.catalog.ImpaladCatalog;
import org.apache.impala.catalog.KuduTable;
-import org.apache.impala.catalog.Table;
import org.apache.impala.catalog.Type;
import org.apache.impala.common.AnalysisException;
import org.apache.impala.common.FileSystemUtil;
@@ -146,6 +145,7 @@ import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.base.Predicates;
import com.google.common.collect.Lists;
@@ -163,39 +163,53 @@ public class Frontend {
// Max time to wait for a catalog update notification.
public static final long MAX_CATALOG_UPDATE_WAIT_TIME_MS = 2 * 1000;
- //TODO: Make the reload interval configurable.
+ // TODO: Make the reload interval configurable.
private static final int AUTHORIZATION_POLICY_RELOAD_INTERVAL_SECS = 5 * 60;
- private final AtomicReference<ImpaladCatalog> impaladCatalog_ =
- new AtomicReference<ImpaladCatalog>();
+ private final FeCatalogManager catalogManager_;
private final AuthorizationConfig authzConfig_;
- private final AtomicReference<AuthorizationChecker> authzChecker_;
+ /**
+ * Authorization checker. Initialized and periodically loaded by a task
+ * running on the {@link #policyReader_} thread.
+ */
+ private final AtomicReference<AuthorizationChecker> authzChecker_ =
+ new AtomicReference<>();
private final ScheduledExecutorService policyReader_ =
Executors.newScheduledThreadPool(1);
- private final String defaultKuduMasterHosts_;
- public Frontend(AuthorizationConfig authorizationConfig,
- String defaultKuduMasterHosts) {
- this(authorizationConfig, new ImpaladCatalog(defaultKuduMasterHosts));
+ public Frontend(AuthorizationConfig authorizationConfig) {
+ this(authorizationConfig, FeCatalogManager.createFromBackendConfig());
}
/**
- * C'tor used by tests to pass in a custom ImpaladCatalog.
+ * Create a frontend with a specific catalog instance which will not allow
+ * updates and will be used for all requests.
*/
- public Frontend(AuthorizationConfig authorizationConfig, ImpaladCatalog catalog) {
+ @VisibleForTesting
+ public Frontend(AuthorizationConfig authorizationConfig,
+ FeCatalog testCatalog) {
+ this(authorizationConfig, FeCatalogManager.createForTests(testCatalog));
+ }
+
+ private Frontend(AuthorizationConfig authorizationConfig,
+ FeCatalogManager catalogManager) {
authzConfig_ = authorizationConfig;
- impaladCatalog_.set(catalog);
- defaultKuduMasterHosts_ = catalog.getDefaultKuduMasterHosts();
- authzChecker_ = new AtomicReference<AuthorizationChecker>(
- new AuthorizationChecker(authzConfig_, impaladCatalog_.get().getAuthPolicy()));
+ catalogManager_ = catalogManager;
+
+ // Load the authorization policy once at startup, initializing
+ // authzChecker_. This ensures that, if the policy fails to load,
+ // we will throw an exception and fail to start.
+ AuthorizationPolicyReader policyReaderTask =
+ new AuthorizationPolicyReader(authzConfig_);
+ policyReaderTask.run();
+
// If authorization is enabled, reload the policy on a regular basis.
if (authzConfig_.isEnabled() && authzConfig_.isFileBasedPolicy()) {
// Stagger the reads across nodes
Random randomGen = new Random(UUID.randomUUID().hashCode());
int delay = AUTHORIZATION_POLICY_RELOAD_INTERVAL_SECS + randomGen.nextInt(60);
- policyReader_.scheduleAtFixedRate(
- new AuthorizationPolicyReader(authzConfig_),
+ policyReader_.scheduleAtFixedRate(policyReaderTask,
delay, AUTHORIZATION_POLICY_RELOAD_INTERVAL_SECS, TimeUnit.SECONDS);
}
}
@@ -222,26 +236,16 @@ public class Frontend {
}
}
- public ImpaladCatalog getCatalog() { return impaladCatalog_.get(); }
+ public FeCatalog getCatalog() { return catalogManager_.getOrCreateCatalog(); }
+
public AuthorizationChecker getAuthzChecker() { return authzChecker_.get(); }
public TUpdateCatalogCacheResponse updateCatalogCache(
TUpdateCatalogCacheRequest req) throws CatalogException, TException {
- if (req.is_delta) return impaladCatalog_.get().updateCatalog(req);
-
- // If this is not a delta, this update should replace the current
- // Catalog contents so create a new catalog and populate it.
- ImpaladCatalog catalog = new ImpaladCatalog(defaultKuduMasterHosts_);
-
- TUpdateCatalogCacheResponse response = catalog.updateCatalog(req);
-
- // Now that the catalog has been updated, replace the references to
- // impaladCatalog_/authzChecker_. This ensures that clients don't see the catalog
- // disappear. The catalog is guaranteed to be ready since updateCatalog() has a
- // postcondition of isReady() == true.
- impaladCatalog_.set(catalog);
- authzChecker_.set(new AuthorizationChecker(authzConfig_, catalog.getAuthPolicy()));
- return response;
+ TUpdateCatalogCacheResponse resp = catalogManager_.updateCatalogCache(req);
+ authzChecker_.set(new AuthorizationChecker(
+ authzConfig_, getCatalog().getAuthPolicy()));
+ return resp;
}
/**
@@ -555,7 +559,7 @@ public class Frontend {
// Get the destination for the load. If the load is targeting a partition,
// this the partition location. Otherwise this is the table location.
String destPathString = null;
- ImpaladCatalog catalog = impaladCatalog_.get();
+ FeCatalog catalog = getCatalog();
if (request.isSetPartition_spec()) {
destPathString = catalog.getHdfsPartition(tableName.getDb(),
tableName.getTbl(), request.getPartition_spec()).getLocation();
@@ -616,7 +620,7 @@ public class Frontend {
*/
public List<String> getTableNames(String dbName, PatternMatcher matcher,
User user) throws ImpalaException {
- List<String> tblNames = impaladCatalog_.get().getTableNames(dbName, matcher);
+ List<String> tblNames = getCatalog().getTableNames(dbName, matcher);
if (authzConfig_.isEnabled()) {
Iterator<String> iter = tblNames.iterator();
while (iter.hasNext()) {
@@ -635,7 +639,7 @@ public class Frontend {
* Returns a list of columns of a table using 'matcher' and are accessible
* to the given user.
*/
- public List<Column> getColumns(Table table, PatternMatcher matcher,
+ public List<Column> getColumns(FeTable table, PatternMatcher matcher,
User user) throws InternalException {
Preconditions.checkNotNull(table);
Preconditions.checkNotNull(matcher);
@@ -660,7 +664,7 @@ public class Frontend {
*/
public List<? extends FeDb> getDbs(PatternMatcher matcher, User user)
throws InternalException {
- List<? extends FeDb> dbs = impaladCatalog_.get().getDbs(matcher);
+ List<? extends FeDb> dbs = getCatalog().getDbs(matcher);
// If authorization is enabled, filter out the databases the user does not
// have permissions on.
if (authzConfig_.isEnabled()) {
@@ -692,7 +696,7 @@ public class Frontend {
* matches all data sources.
*/
public List<? extends FeDataSource> getDataSrcs(String pattern) {
- return impaladCatalog_.get().getDataSources(
+ return getCatalog().getDataSources(
PatternMatcher.createHivePatternMatcher(pattern));
}
@@ -701,7 +705,7 @@ public class Frontend {
*/
public TResultSet getColumnStats(String dbName, String tableName)
throws ImpalaException {
- Table table = impaladCatalog_.get().getTable(dbName, tableName);
+ FeTable table = getCatalog().getTable(dbName, tableName);
TResultSet result = new TResultSet();
TResultSetMetadata resultSchema = new TResultSetMetadata();
result.setSchema(resultSchema);
@@ -729,7 +733,7 @@ public class Frontend {
*/
public TResultSet getTableStats(String dbName, String tableName, TShowStatsOp op)
throws ImpalaException {
- Table table = impaladCatalog_.get().getTable(dbName, tableName);
+ FeTable table = getCatalog().getTable(dbName, tableName);
if (table instanceof FeFsTable) {
return ((FeFsTable) table).getTableStats();
} else if (table instanceof HBaseTable) {
@@ -755,7 +759,7 @@ public class Frontend {
public List<Function> getFunctions(TFunctionCategory category,
String dbName, String fnPattern, boolean exactMatch)
throws DatabaseNotFoundException {
- Db db = impaladCatalog_.get().getDb(dbName);
+ FeDb db = getCatalog().getDb(dbName);
if (db == null) {
throw new DatabaseNotFoundException("Database '" + dbName + "' not found");
}
@@ -783,7 +787,7 @@ public class Frontend {
*/
public TDescribeResult describeDb(String dbName, TDescribeOutputStyle outputStyle)
throws ImpalaException {
- Db db = impaladCatalog_.get().getDb(dbName);
+ FeDb db = getCatalog().getDb(dbName);
return DescribeResultFactory.buildDescribeDbResult(db, outputStyle);
}
@@ -794,7 +798,7 @@ public class Frontend {
*/
public TDescribeResult describeTable(TTableName tableName,
TDescribeOutputStyle outputStyle, User user) throws ImpalaException {
- Table table = impaladCatalog_.get().getTable(tableName.db_name, tableName.table_name);
+ FeTable table = getCatalog().getTable(tableName.db_name, tableName.table_name);
List<Column> filteredColumns;
if (authzConfig_.isEnabled()) {
// First run a table check
@@ -1120,14 +1124,14 @@ public class Frontend {
// create finalization params of insert stmt
InsertStmt insertStmt = analysisResult.getInsertStmt();
- if (insertStmt.getTargetTable() instanceof HdfsTable) {
+ if (insertStmt.getTargetTable() instanceof FeFsTable) {
TFinalizeParams finalizeParams = new TFinalizeParams();
finalizeParams.setIs_overwrite(insertStmt.isOverwrite());
finalizeParams.setTable_name(insertStmt.getTargetTableName().getTbl());
finalizeParams.setTable_id(DescriptorTable.TABLE_SINK_ID);
String db = insertStmt.getTargetTableName().getDb();
finalizeParams.setTable_db(db == null ? queryCtx.session.database : db);
- HdfsTable hdfsTable = (HdfsTable) insertStmt.getTargetTable();
+ FeFsTable hdfsTable = (FeFsTable) insertStmt.getTargetTable();
finalizeParams.setHdfs_base_dir(hdfsTable.getHdfsBaseDir());
finalizeParams.setStaging_dir(
hdfsTable.getHdfsBaseDir() + "/_impala_insert_staging");
@@ -1208,10 +1212,10 @@ public class Frontend {
*/
public TResultSet getTableFiles(TShowFilesParams request)
throws ImpalaException{
- Table table = impaladCatalog_.get().getTable(request.getTable_name().getDb_name(),
+ FeTable table = getCatalog().getTable(request.getTable_name().getDb_name(),
request.getTable_name().getTable_name());
- if (table instanceof HdfsTable) {
- return ((HdfsTable) table).getFiles(request.getPartition_set());
+ if (table instanceof FeFsTable) {
+ return ((FeFsTable) table).getFiles(request.getPartition_set());
} else {
throw new InternalException("SHOW FILES only supports Hdfs table. " +
"Unsupported table class: " + table.getClass());
diff --git a/fe/src/main/java/org/apache/impala/service/JniFrontend.java b/fe/src/main/java/org/apache/impala/service/JniFrontend.java
index d8ef912..25dee1c 100644
--- a/fe/src/main/java/org/apache/impala/service/JniFrontend.java
+++ b/fe/src/main/java/org/apache/impala/service/JniFrontend.java
@@ -148,7 +148,7 @@ public class JniFrontend {
}
LOG.info(JniUtil.getJavaVersion());
- frontend_ = new Frontend(authConfig, cfg.kudu_master_hosts);
+ frontend_ = new Frontend(authConfig);
}
/**
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 ea4864f..d221e2c 100644
--- a/fe/src/main/java/org/apache/impala/service/MetadataOp.java
+++ b/fe/src/main/java/org/apache/impala/service/MetadataOp.java
@@ -28,12 +28,12 @@ import org.apache.impala.analysis.TableName;
import org.apache.impala.authorization.User;
import org.apache.impala.catalog.Catalog;
import org.apache.impala.catalog.Column;
+import org.apache.impala.catalog.FeCatalog;
import org.apache.impala.catalog.FeDb;
+import org.apache.impala.catalog.FeTable;
import org.apache.impala.catalog.Function;
-import org.apache.impala.catalog.ImpaladCatalog;
import org.apache.impala.catalog.PrimitiveType;
import org.apache.impala.catalog.ScalarType;
-import org.apache.impala.catalog.Table;
import org.apache.impala.catalog.Type;
import org.apache.impala.common.ImpalaException;
import org.apache.impala.thrift.TColumn;
@@ -266,7 +266,7 @@ public class MetadataOp {
return result;
}
- ImpaladCatalog catalog = fe.getCatalog();
+ FeCatalog catalog = fe.getCatalog();
for (FeDb db: fe.getDbs(schemaPatternMatcher, user)) {
if (fnPatternMatcher != PatternMatcher.MATCHER_MATCH_NONE) {
// Get function metadata
@@ -279,7 +279,7 @@ public class MetadataOp {
List<String> tableComments = Lists.newArrayList();
List<String> tableTypes = Lists.newArrayList();
for (String tabName: fe.getTableNames(db.getName(), tablePatternMatcher, user)) {
- Table table = catalog.getTable(db.getName(), tabName);
+ FeTable table = catalog.getTable(db.getName(), tabName);
if (table == null) continue;
String comment = null;