You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by st...@apache.org on 2020/06/10 17:46:59 UTC
[impala] 04/04: IMPALA-9818: Add fetch size as option to impala
shell
This is an automated email from the ASF dual-hosted git repository.
stakiar pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 3088ca8580ed2f8c87235cbee3f3b3a90c292218
Author: Sahil Takiar <ta...@gmail.com>
AuthorDate: Fri Jun 5 13:32:54 2020 -0700
IMPALA-9818: Add fetch size as option to impala shell
Adds the option --fetch_size to the Impala shell. This new option allows
users to specify the fetch size used when issuing fetch RPCs to the
Impala Coordinator (e.g. TFetchResultsReq and BeeswaxService.fetch).
This parameter applies for all client protocols: beeswax, hs2, hs2-http.
The default --fetch_size is set to 10240 (10x the default batch size).
The new --fetch_size parameter is most effective when result spooling is
enabled. When result spooling is disabled, Impala can only return a
single row batch per fetch RPC (so 1024 rows by default). When result
spooling is enabled, Impala can return up to 100 row batches per fetch
request.
Removes some logic in the the impala_client.py file that attempts to
simulate a fetch_size. The code would issue multiple fetch requests to
fullfill the given fetch_size. This logic is no longer needed now that
result spooling is available.
Testing:
* Ran core tests
* Added new tests in test_shell_client.py and test_shell_commandline.py
Change-Id: I8dc7962aada6b38795241d067a99bd94fabca57b
Reviewed-on: http://gerrit.cloudera.org:8080/16041
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Sahil Takiar <st...@cloudera.com>
---
shell/impala_client.py | 31 ++------
shell/impala_shell.py | 25 +++---
shell/option_parser.py | 13 ++++
tests/custom_cluster/test_hs2_fault_injection.py | 4 +-
tests/shell/test_shell_client.py | 96 ++++++++++++++++++++++++
tests/shell/test_shell_commandline.py | 9 +++
6 files changed, 141 insertions(+), 37 deletions(-)
diff --git a/shell/impala_client.py b/shell/impala_client.py
index 9e5704b..8973d59 100755
--- a/shell/impala_client.py
+++ b/shell/impala_client.py
@@ -114,7 +114,7 @@ class ImpalaClient(object):
"""Base class for shared functionality between HS2 and Beeswax. Includes stub methods
for methods that are expected to be implemented in the subclasses.
TODO: when beeswax support is removed, merge this with ImpalaHS2Client."""
- def __init__(self, impalad, kerberos_host_fqdn, use_kerberos=False,
+ def __init__(self, impalad, fetch_size, kerberos_host_fqdn, use_kerberos=False,
kerberos_service_name="impala", use_ssl=False, ca_cert=None, user=None,
ldap_password=None, use_ldap=False, client_connect_timeout_ms=60000,
verbose=True, use_http_base_transport=False, http_path=None):
@@ -133,7 +133,7 @@ class ImpalaClient(object):
self.client_connect_timeout_ms = int(client_connect_timeout_ms)
self.default_query_options = {}
self.query_option_levels = {}
- self.fetch_batch_size = 1024
+ self.fetch_size = fetch_size
self.use_http_base_transport = use_http_base_transport
self.http_path = http_path
# This is set from ImpalaShell's signal handler when a query is cancelled
@@ -264,25 +264,10 @@ class ImpalaClient(object):
def fetch(self, query_handle):
"""Returns an iterable of batches of result rows. Each batch is an iterable of rows.
Each row is an iterable of strings in the format in which they should be displayed
- Tries to ensure that the batches have a granularity of self.fetch_batch_size but
+ Tries to ensure that the batches have a granularity of self.fetch_size but
does not guarantee it.
"""
- result_rows = None
- for rows in self._fetch_one_batch(query_handle):
- if result_rows:
- result_rows.extend(rows)
- else:
- result_rows = rows
- if len(result_rows) > self.fetch_batch_size:
- yield result_rows
- result_rows = None
-
- # Return the final batch of rows.
- if result_rows:
- yield result_rows
-
- def _fetch_one_batch(self, query_handle):
- """Returns an iterable of batches of result rows up to self.fetch_batch_size. Does
+ """Returns an iterable of batches of result rows up to self.fetch_size. Does
not need to consolidate those batches into larger batches."""
raise NotImplementedError()
@@ -761,7 +746,7 @@ class ImpalaHS2Client(ImpalaClient):
return "{low}:{high}".format(low=low_hex, high=high_hex)
- def _fetch_one_batch(self, query_handle):
+ def fetch(self, query_handle):
assert query_handle.hasResultSet
prim_types = [column.typeDesc.types[0].primitiveEntry.type
for column in query_handle.schema.columns]
@@ -769,7 +754,7 @@ class ImpalaHS2Client(ImpalaClient):
for prim_type in prim_types]
while True:
req = TFetchResultsReq(query_handle, TFetchOrientation.FETCH_NEXT,
- self.fetch_batch_size)
+ self.fetch_size)
def FetchResults():
return self.imp_service.FetchResults(req)
@@ -1083,11 +1068,11 @@ class ImpalaBeeswaxClient(ImpalaClient):
return self.ERROR_STATE
return state
- def _fetch_one_batch(self, query_handle):
+ def fetch(self, query_handle):
while True:
result, rpc_status = self._do_beeswax_rpc(
lambda: self.imp_service.fetch(query_handle, False,
- self.fetch_batch_size))
+ self.fetch_size))
if rpc_status != RpcStatus.OK:
raise RPCException()
yield [row.split('\t') for row in result.data]
diff --git a/shell/impala_shell.py b/shell/impala_shell.py
index 322d94b..244bef2 100755
--- a/shell/impala_shell.py
+++ b/shell/impala_shell.py
@@ -217,6 +217,7 @@ class ImpalaShell(cmd.Cmd, object):
self.ignore_query_failure = options.ignore_query_failure
self.http_path = options.http_path
+ self.fetch_size = options.fetch_size
# Due to a readline bug in centos/rhel7, importing it causes control characters to be
# printed. This breaks any scripting against the shell in non-interactive mode. Since
@@ -543,22 +544,22 @@ class ImpalaShell(cmd.Cmd, object):
def _new_impala_client(self):
protocol = options.protocol.lower()
if protocol == 'hs2':
- return ImpalaHS2Client(self.impalad, self.kerberos_host_fqdn, self.use_kerberos,
- self.kerberos_service_name, self.use_ssl,
- self.ca_cert, self.user, self.ldap_password,
- self.use_ldap, self.client_connect_timeout_ms, self.verbose,
+ return ImpalaHS2Client(self.impalad, self.fetch_size, self.kerberos_host_fqdn,
+ self.use_kerberos, self.kerberos_service_name, self.use_ssl,
+ self.ca_cert, self.user, self.ldap_password, self.use_ldap,
+ self.client_connect_timeout_ms, self.verbose,
use_http_base_transport=False, http_path=self.http_path)
elif protocol == 'hs2-http':
- return ImpalaHS2Client(self.impalad, self.kerberos_host_fqdn, self.use_kerberos,
- self.kerberos_service_name, self.use_ssl,
- self.ca_cert, self.user, self.ldap_password,
- self.use_ldap, self.client_connect_timeout_ms, self.verbose,
+ return ImpalaHS2Client(self.impalad, self.fetch_size, self.kerberos_host_fqdn,
+ self.use_kerberos, self.kerberos_service_name, self.use_ssl,
+ self.ca_cert, self.user, self.ldap_password, self.use_ldap,
+ self.client_connect_timeout_ms, self.verbose,
use_http_base_transport=True, http_path=self.http_path)
elif protocol == 'beeswax':
- return ImpalaBeeswaxClient(self.impalad, self.kerberos_host_fqdn, self.use_kerberos,
- self.kerberos_service_name, self.use_ssl,
- self.ca_cert, self.user, self.ldap_password,
- self.use_ldap, self.client_connect_timeout_ms, self.verbose)
+ return ImpalaBeeswaxClient(self.impalad, self.fetch_size, self.kerberos_host_fqdn,
+ self.use_kerberos, self.kerberos_service_name, self.use_ssl,
+ self.ca_cert, self.user, self.ldap_password, self.use_ldap,
+ self.client_connect_timeout_ms, self.verbose)
else:
err_msg = "Invalid --protocol value {0}, must be beeswax, hs2 or hs2-http."
print(err_msg.format(protocol), file=sys.stderr)
diff --git a/shell/option_parser.py b/shell/option_parser.py
index 3a4f090..a48c053 100755
--- a/shell/option_parser.py
+++ b/shell/option_parser.py
@@ -284,6 +284,19 @@ def get_option_parser(defaults):
"enforce any http path for the incoming requests, deployments could "
"still put it behind a loadbalancer that can expect the traffic at a "
"certain path.")
+ parser.add_option("--fetch_size", type="int", dest="fetch_size", default=10240,
+ help="The fetch size when fetching rows from the Impala coordinator. "
+ "The fetch size controls how many rows a single fetch RPC request "
+ "(RPC from the Impala shell to the Impala coordinator) reads at a "
+ "time. This option is most effective when result spooling is enabled "
+ "('spool_query_results'=true). When result spooling is enabled "
+ "values over the batch_size are honored. When result spooling is "
+ "disabled, values over the batch_size have no affect. By default, "
+ "the fetch_size is set to 10240 which is equivalent to 10 row "
+ "batches (assuming the default batch size). Note that if result "
+ "spooling is disabled only a single row batch can be fetched at a "
+ "time regardless of the specified fetch_size.")
+
# add default values to the help text
for option in parser.option_list:
diff --git a/tests/custom_cluster/test_hs2_fault_injection.py b/tests/custom_cluster/test_hs2_fault_injection.py
index 2393cf4..f450287 100644
--- a/tests/custom_cluster/test_hs2_fault_injection.py
+++ b/tests/custom_cluster/test_hs2_fault_injection.py
@@ -106,7 +106,7 @@ class TestHS2FaultInjection(CustomClusterTestSuite):
impala-shell client"""
def setup(self):
impalad = IMPALAD_HS2_HTTP_HOST_PORT.split(":")
- self.custom_hs2_http_client = FaultInjectingImpalaHS2Client(impalad,
+ self.custom_hs2_http_client = FaultInjectingImpalaHS2Client(impalad, 1024,
kerberos_host_fqdn=None, use_http_base_transport=True, http_path='cliservice')
self.transport = self.custom_hs2_http_client.transport
@@ -317,7 +317,7 @@ class TestHS2FaultInjection(CustomClusterTestSuite):
self.transport.close()
rows_fetched = self.custom_hs2_http_client.fetch(query_handle)
for rows in rows_fetched:
- num_rows += 1
+ num_rows += len(rows)
assert num_rows == 1
self.transport.close()
self.close_query(query_handle)
diff --git a/tests/shell/test_shell_client.py b/tests/shell/test_shell_client.py
new file mode 100644
index 0000000..c5321eb
--- /dev/null
+++ b/tests/shell/test_shell_client.py
@@ -0,0 +1,96 @@
+#!/usr/bin/env impala-python
+# -*- coding: utf-8 -*-
+#
+# 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.
+
+from shell.impala_client import ImpalaBeeswaxClient, ImpalaHS2Client
+from tests.common.impala_test_suite import ImpalaTestSuite
+from tests.common.test_dimensions import create_client_protocol_dimension
+from util import get_impalad_host_port
+
+
+class TestShellClient(ImpalaTestSuite):
+ """Tests for the Impala Shell clients: ImpalaBeeswaxClient and ImpalaHS2Client."""
+
+ @classmethod
+ def get_workload(self):
+ return 'functional-query'
+
+ @classmethod
+ def add_test_dimensions(cls):
+ cls.ImpalaTestMatrix.add_dimension(create_client_protocol_dimension())
+
+ def test_fetch_size(self, vector):
+ """Tests that when result spooling is disabled, setting a small batch_size causes
+ the shell to fetch a single batch at a time, even when the configured fetch size is
+ larger than the batch_size."""
+ handle = None
+ num_rows = 100
+ batch_size = 10
+ query_options = {'batch_size': str(batch_size)}
+ client = self.__get_shell_client(vector)
+
+ try:
+ client.connect()
+ handle = client.execute_query(
+ "select * from functional.alltypes limit {0}".format(num_rows), query_options)
+ self.__fetch_rows(client.fetch(handle), batch_size, num_rows)
+ finally:
+ if handle is not None: client.close_query(handle)
+ client.close_connection()
+
+ def test_fetch_size_result_spooling(self, vector):
+ """Tests that when result spooling is enabled, that the exact fetch_size is honored
+ even if a small batch_size is configured."""
+ handle = None
+ fetch_size = 20
+ num_rows = 100
+ query_options = {'batch_size': '10', 'spool_query_results': 'true'}
+ client = self.__get_shell_client(vector, fetch_size)
+
+ try:
+ client.connect()
+ handle = client.execute_query(
+ "select * from functional.alltypes limit {0}".format(num_rows), query_options)
+ self.__fetch_rows(client.fetch(handle), num_rows / fetch_size, num_rows)
+ finally:
+ if handle is not None: client.close_query(handle)
+ client.close_connection()
+
+ def __fetch_rows(self, fetch_batches, num_batches, num_rows):
+ """Fetches all rows using the given fetch_batches generator. Asserts that num_batches
+ batches are produced by the generator and that num_rows are returned."""
+ num_batches_count = 0
+ rows_per_batch = num_rows / num_batches
+ for fetch_batch in fetch_batches:
+ assert len(fetch_batch) == rows_per_batch
+ num_batches_count += 1
+ if num_batches_count == num_batches: break
+ assert num_batches_count == num_batches
+
+ def __get_shell_client(self, vector, fetch_size=1024):
+ """Returns the client specified by the protocol in the given vector."""
+ impalad = get_impalad_host_port(vector).split(":")
+ protocol = vector.get_value("protocol")
+ if protocol == 'hs2':
+ return ImpalaHS2Client(impalad, fetch_size, None)
+ elif protocol == 'hs2-http':
+ return ImpalaHS2Client(impalad, fetch_size, None,
+ use_http_base_transport=True, http_path='cliservice')
+ elif protocol == 'beeswax':
+ return ImpalaBeeswaxClient(impalad, fetch_size, None)
diff --git a/tests/shell/test_shell_commandline.py b/tests/shell/test_shell_commandline.py
index b2fcb59..34d9dd9 100644
--- a/tests/shell/test_shell_commandline.py
+++ b/tests/shell/test_shell_commandline.py
@@ -1017,3 +1017,12 @@ class TestImpalaShell(ImpalaTestSuite):
assert "3\t3\t30.3" in result.stdout, result.stdout
assert "4\t4\t40.4" in result.stdout, result.stdout
+
+ def test_fetch_size(self, vector):
+ """Test the --fetch_size option with and without result spooling enabled."""
+ query = "select * from functional.alltypes limit 1024"
+ query_with_result_spooling = "set spool_query_results=true; " + query
+ for query in [query, query_with_result_spooling]:
+ result = run_impala_shell_cmd(vector, ['-q', query, '-B', '--fetch_size', '512'])
+ result_rows = result.stdout.strip().split('\n')
+ assert len(result_rows) == 1024