You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by kw...@apache.org on 2018/09/05 20:12:24 UTC
[1/9] impala git commit: IMPALA-6923: Remove create_database.py and
perf_result_datastore.py
Repository: impala
Updated Branches:
refs/heads/master d4e281b73 -> 3433d8fe1
IMPALA-6923: Remove create_database.py and perf_result_datastore.py
Description:
The current upstream usage indicates that report_benchmark_results.py script
is the only script in this folder that is used to generate a report comparing
performance between two given runs of the performance tests. The other two
scripts create_database.py and perf_result_datastore.py store some metrics
from a performance test run to database on a specified impala instance.
However they rely on some Internal resources not available to
external apache community to generate a meaningful interpretation/report of the
metrics stored in the performance database.
This change also removes lines of code from report_benchmark_results.py
that were previously parsing command line options, that were being used by the
two scripts that are being removed.
Change-Id: I2c9b04f0d0e8ca221e383aab2ba3ee53d968dd10
Reviewed-on: http://gerrit.cloudera.org:8080/11315
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/impala/commit/37c66996
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/37c66996
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/37c66996
Branch: refs/heads/master
Commit: 37c66996a0819a81d8b209f73b52fd2c4c99129a
Parents: d4e281b
Author: njanarthanan <nj...@cloudera.com>
Authored: Fri Aug 24 10:03:48 2018 -0700
Committer: Impala Public Jenkins <im...@cloudera.com>
Committed: Fri Aug 31 23:47:44 2018 +0000
----------------------------------------------------------------------
tests/benchmark/create_database.py | 98 ---------
tests/benchmark/perf_result_datastore.py | 264 -----------------------
tests/benchmark/report_benchmark_results.py | 70 ++----
3 files changed, 15 insertions(+), 417 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/37c66996/tests/benchmark/create_database.py
----------------------------------------------------------------------
diff --git a/tests/benchmark/create_database.py b/tests/benchmark/create_database.py
deleted file mode 100755
index a38c635..0000000
--- a/tests/benchmark/create_database.py
+++ /dev/null
@@ -1,98 +0,0 @@
-#!/usr/bin/env impala-python
-#
-# 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 impala.dbapi import connect as impala_connect
-from optparse import OptionParser
-
-parser = OptionParser()
-parser.add_option('--host')
-parser.add_option('--port', type='int', default=21050)
-parser.add_option('--db_name', default='impala_perf_results')
-parser.add_option('--use_secure_connection', default=False, action='store_true')
-
-options, args = parser.parse_args()
-
-CREATE_STATEMENTS = ['''
- CREATE EXTERNAL TABLE IF NOT EXISTS ExecutionResults (
- result_id string,
- run_info_id string,
- query_id string,
- workload_id string,
- file_type_id string,
- num_clients INT,
- num_iterations INT,
- cluster_name string,
- executor_name string,
- exec_time double,
- run_date timestamp,
- version string
-) stored as parquet;
-''',
-'''
-CREATE EXTERNAL TABLE IF NOT EXISTS Queries (
- query_id string,
- name string,
- query_string string,
- notes string
-) stored as parquet;
-''',
-'''
-CREATE EXTERNAL TABLE IF NOT EXISTS Workloads (
- workload_id string,
- name string,
- scale_factor string
-) stored as parquet;
-''',
-'''
-CREATE EXTERNAL TABLE IF NOT EXISTS FileTypes (
- file_type_id string,
- file_format string,
- compression_codec string,
- compression_type string
-) stored as parquet;
-''',
-'''
-CREATE EXTERNAL TABLE IF NOT EXISTS RunInfos (
- run_info_id string,
- run_info string,
- user string
-) stored as parquet;
-''',
-'''
-CREATE EXTERNAL TABLE IF NOT EXISTS RuntimeProfiles (
- result_id string,
- profile string
-) stored as parquet;
-''']
-
-def create_database():
- if options.use_secure_connection:
- connection = impala_connect(options.host, options.port,
- use_ssl=True, auth_mechanism='GSSAPI')
- else:
- connection = impala_connect(options.host, options.port)
- cursor = connection.cursor()
-
- cursor.execute('CREATE DATABASE IF NOT EXISTS {0}'.format(options.db_name))
- cursor.execute('USE {0}'.format(options.db_name))
- for create_statement in CREATE_STATEMENTS:
- cursor.execute(create_statement)
-
-if __name__ == '__main__':
- create_database()
http://git-wip-us.apache.org/repos/asf/impala/blob/37c66996/tests/benchmark/perf_result_datastore.py
----------------------------------------------------------------------
diff --git a/tests/benchmark/perf_result_datastore.py b/tests/benchmark/perf_result_datastore.py
deleted file mode 100755
index 1eb4ae0..0000000
--- a/tests/benchmark/perf_result_datastore.py
+++ /dev/null
@@ -1,264 +0,0 @@
-# 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.
-
-import logging
-import uuid
-from functools import wraps
-from impala.dbapi import connect as impala_connect
-from tests.util.calculation_util import get_random_id
-
-LOG = logging.getLogger(__name__)
-
-class PerfResultDataStore(object):
- '''Allows for interaction with the performance Impala database.
-
- In order to reduce the number of calls to the database we cache query_id, workload_id,
- file_type_id and run_info_id in _cache. If an id cannot be found in the cache, we make a
- request to the database. Objects not in the database will be automatically inserted.
-
- In order to avoid creating a new file per query, we create a staging table for execution
- results and query profiles. When all results have been inserted, we insert the data from
- the staging tables into ExecutionResults and RuntimeProfiles. This way, we create on the
- order of 1 parquet file per run.'''
-
- def __init__(self, host, port, database_name, use_secure_connection=False):
- if use_secure_connection:
- self._connection = impala_connect(host, port, use_ssl=True, auth_mechanism='GSSAPI')
- else:
- self._connection = impala_connect(host, port)
- self._database_name = database_name
- self._staging_exec_result_table = None
- self._staging_profile_table = None
- self._cache = {}
-
- def __enter__(self):
- self._staging_exec_result_table = 'ExecutionResultsStaging_' + get_random_id(5)
- self._staging_profile_table = 'RuntimeProfilesStaging_' + get_random_id(5)
- self._create_new_table_as('ExecutionResults', self._staging_exec_result_table)
- self._create_new_table_as('RuntimeProfiles', self._staging_profile_table)
- return self
-
- def __exit__(self, type, value, traceback):
- self._copy_to_table(self._staging_exec_result_table, 'ExecutionResults')
- self._copy_to_table(self._staging_profile_table, 'RuntimeProfiles')
- self._drop_table(self._staging_exec_result_table)
- self._drop_table(self._staging_profile_table)
-
- def insert_execution_result(self,
- query_name,
- query_string,
- workload_name,
- scale_factor,
- file_format,
- compression_codec,
- compression_type,
- num_clients,
- num_iterations,
- cluster_name,
- executor_name,
- exec_time,
- run_date,
- version,
- run_info,
- user_name,
- runtime_profile):
-
- LOG.debug('About to insert execution result.')
-
- query_id = self._get_query_id(query_name, query_string)
- workload_id = self._get_workload_id(workload_name, scale_factor)
- file_type_id = self._get_file_type_id(
- file_format, compression_codec, compression_type)
- run_info_id = self._get_run_info_id(run_info, user_name)
-
- result_id = self._insert_execution_result(
- run_info_id, query_id, workload_id, file_type_id,
- num_clients, num_iterations, cluster_name, executor_name,
- exec_time, run_date, version)
-
- self._insert_runtime_profile(result_id, runtime_profile)
-
- def _get_query_id(self, query_name, query_string):
- ''' Gets the query_id for the given query name and query text. If not found in the
- database, this information gets inserted into the database. Results are cached. '''
- if (query_name, query_string) not in self._cache:
- self._cache[(query_name, query_string)] = \
- self._get_query_id_remote(query_name, query_string) or \
- self._insert_query_info(query_name, query_string, '')
- # Empty string above represents notes. No notes will be inserted by default.
- return self._cache[(query_name, query_string)]
-
- def _get_workload_id(self, workload_name, scale_factor):
- ''' Gets the workload_id for the given workload / scale factor. If not found in the
- database, this information gets inserted into the database. Results are cached.'''
- if (workload_name, scale_factor) not in self._cache:
- self._cache[(workload_name, scale_factor)] = \
- self._get_workload_id_remote(workload_name, scale_factor) or \
- self._insert_workload_info(workload_name, scale_factor)
- return self._cache[(workload_name, scale_factor)]
-
- def _get_file_type_id(self, file_format, compression_codec, compression_type):
- ''' Gets the file_type_id for the fiven file_format / compression codec / compression
- type combination. If not found in the database, this information gets inserted into
- the database. Results are cached.'''
- if (file_format, compression_codec, compression_type) not in self._cache:
- self._cache[(file_format, compression_codec, compression_type)] = \
- self._get_file_type_id_remote(
- file_format, compression_codec, compression_type) or \
- self._insert_filetype_info(file_format, compression_codec, compression_type)
- return self._cache[(file_format, compression_codec, compression_type)]
-
- def _get_run_info_id(self, run_info, user):
- '''Gets the run_info id for the given run_info and user. Creates a new entry for each
- (run_info, user) pair. Results are cached.'''
- if (run_info, user) not in self._cache:
- self._cache[(run_info, user)] = self._insert_run_info(run_info, user)
- return self._cache[(run_info, user)]
-
- def _cursor_wrapper(function):
- ''' Handles the common initialize/close pattern for cursor objects '''
- @wraps(function)
- def wrapper(self, *args, **kwargs):
- cursor = self._connection.cursor()
- cursor.execute('use {0}'.format(self._database_name))
- result = function(self, *args, cursor=cursor)
- cursor.close()
- return result
- return wrapper
-
- @_cursor_wrapper
- def _get_query_id_remote(self, query_name, query_string, cursor):
- cursor.execute('''
- select query_id
- from Queries where name="{0}"
- and query_string="{1}"'''.format(query_name, query_string))
- query_id = cursor.fetchone()
- return query_id[0] if query_id else None
-
- @_cursor_wrapper
- def _get_workload_id_remote(self, workload, scale_factor, cursor):
- cursor.execute('''
- select workload_id
- from Workloads where name="{0}"
- and scale_factor="{1}"'''.format(workload, scale_factor))
- workload_id = cursor.fetchone()
- return workload_id[0] if workload_id else None
-
- @_cursor_wrapper
- def _get_file_type_id_remote(self,
- file_format,
- compression_codec,
- compression_type,
- cursor):
- cursor.execute('''
- select file_type_id
- from FileTypes where file_format="{0}"
- and compression_codec="{1}" and compression_type="{2}"
- '''.format(file_format,compression_codec, compression_type))
- file_type_id = cursor.fetchone()
- return file_type_id[0] if file_type_id else None
-
- @_cursor_wrapper
- def _get_run_info_id_remote(self, run_info, user, cursor):
- cursor.execute('''
- select run_info_id
- from RunInfos where run_info="{0}" and user="{1}"'''.format(run_info, user))
- run_info_id = cursor.fetchone()
- return run_info_id[0] if run_info_id else None
-
- @_cursor_wrapper
- def _create_new_table_as(self, old_table, new_table, cursor):
- cursor.execute('create table {0} like {1}'.format(new_table, old_table))
-
- @_cursor_wrapper
- def _copy_to_table(self, source_table, destination_table, cursor):
- cursor.execute('insert into table {0} select * from {1}'.format(
- destination_table, source_table))
-
- @_cursor_wrapper
- def _drop_table(self, table, cursor):
- cursor.execute('drop table {0}'.format(table))
-
- @_cursor_wrapper
- def _insert_execution_result(self,
- run_info_id, query_id, workload_id, file_type_id,
- num_clients, num_iterations, cluster_name, executor_name,
- exec_time, run_date, version, cursor):
- result_id = str(uuid.uuid1())
- insert_query = '''insert into {0} (result_id,
- run_info_id, query_id, workload_id, file_type_id,
- num_clients, num_iterations, cluster_name, executor_name,
- exec_time, run_date, version)
- values("{1}",
- "{2}", "{3}", "{4}", "{5}",
- {6}, {7}, "{8}", "{9}",
- {10}, cast("{11}" as timestamp), "{12}")
- '''.format(self._staging_exec_result_table, result_id,
- run_info_id, query_id, workload_id, file_type_id,
- num_clients, num_iterations, cluster_name, executor_name,
- exec_time, run_date, version)
- cursor.execute(insert_query)
- return result_id
-
- @_cursor_wrapper
- def _insert_query_info(self, name, query_string, notes, cursor):
- LOG.debug('Inserting Query: {0}'.format(name))
- query_id = str(uuid.uuid1())
- insert_statement = '''
- insert into Queries (query_id, name, query_string, notes)
- values("{0}", "{1}", "{2}", "{3}")'''.format(
- query_id, name, query_string, notes)
- cursor.execute(insert_statement)
- return query_id
-
- @_cursor_wrapper
- def _insert_workload_info(self, name, scale_factor, cursor):
- LOG.debug('Inserting Workload: {0}, {1}'.format(name, scale_factor))
- workload_id = str(uuid.uuid1())
- cursor.execute('''
- insert into Workloads (workload_id, name, scale_factor)
- values("{0}", "{1}", "{2}")'''.format(workload_id, name, scale_factor))
- return workload_id
-
- @_cursor_wrapper
- def _insert_filetype_info(
- self, file_format, compression_codec, compression_type, cursor):
- LOG.debug('Inserting File Type: {0}, {1}, {2}'.format(
- file_format, compression_codec, compression_type))
- file_type_id = str(uuid.uuid1())
- cursor.execute('''
- insert into FileTypes
- (file_type_id, file_format, compression_codec, compression_type)
- values("{0}", "{1}", "{2}", "{3}")'''.format(
- file_type_id, file_format, compression_codec, compression_type))
- return file_type_id
-
- @_cursor_wrapper
- def _insert_run_info(self, run_info, user, cursor):
- LOG.debug('Inserting Run Info: {0}, User: {1}'.format(run_info, user))
- run_info_id = str(uuid.uuid1())
- cursor.execute('''
- insert into RunInfos (run_info_id, run_info, user)
- values ("{0}", "{1}", "{2}")'''.format(run_info_id, run_info, user))
- return run_info_id
-
- @_cursor_wrapper
- def _insert_runtime_profile(self, result_id, profile, cursor):
- LOG.debug('Inserting Runtime Profile')
- cursor.execute('''
- insert into {0} (result_id, profile)
- values ("{1}", "{2}")'''.format(self._staging_profile_table, result_id, profile))
http://git-wip-us.apache.org/repos/asf/impala/blob/37c66996/tests/benchmark/report_benchmark_results.py
----------------------------------------------------------------------
diff --git a/tests/benchmark/report_benchmark_results.py b/tests/benchmark/report_benchmark_results.py
index 286e4a1..fb41965 100755
--- a/tests/benchmark/report_benchmark_results.py
+++ b/tests/benchmark/report_benchmark_results.py
@@ -27,8 +27,8 @@
# By default in Python if you divide an int by another int (5 / 2), the result will also
# be an int (2). The following line changes this behavior so that float will be returned
# if necessary (2.5).
-from __future__ import division
+from __future__ import division
import difflib
import json
import logging
@@ -130,22 +130,9 @@ parser.add_option("--allowed_latency_diff_secs",
dest="allowed_latency_diff_secs", default=0.0, type="float",
help="If specified, only a timing change that differs by more than\
this value will be considered significant.")
-
-# These parameters are specific to recording results in a database. This is optional
-parser.add_option("--save_to_db", dest="save_to_db", action="store_true",
- default=False, help='Saves results to the specified database.')
-parser.add_option("--is_official", dest="is_official", action="store_true",
- default=False, help='Indicates this is an official perf run result')
-parser.add_option("--db_host", dest="db_host", default='localhost',
- help="Machine hosting the database")
-parser.add_option("--db_port", dest="db_port", default='21050',
- help="Port on the machine hosting the database")
-parser.add_option("--db_name", dest="db_name", default='impala_perf_results',
- help="Name of the perf database.")
-parser.add_option("--use_secure_connection", dest="use_secure_connection",
- action='store_true', default=False, help="Connect using ssl + Kerberos.")
options, args = parser.parse_args()
+
def get_dict_from_json(filename):
"""Given a JSON file, return a nested dictionary.
@@ -728,6 +715,8 @@ class CombinedExecSummaries(object):
return False, err + 'no exec summaries Found'
first_exec_summary = exec_summaries[0]
+
+ # This check is for Metadata queries which don't have summaries
if len(first_exec_summary) < 1:
return False, err + 'exec summary contains no nodes'
@@ -742,6 +731,7 @@ class CombinedExecSummaries(object):
return True, str()
+
class ExecSummaryComparison(object):
"""Represents a comparison between two CombinedExecSummaries.
@@ -1041,18 +1031,23 @@ def save_runtime_diffs(results, ref_results, change_significant, is_regression):
with open(runtime_profile_file_path, 'w+') as f:
f.write(runtime_profile_diff)
-def build_exec_summary_str(results, ref_results, for_variability = False):
+
+def build_exec_summary_str(results, ref_results, for_variability=False):
# There is no summary available for Hive after query execution
- if options.hive_results:
- return ""
+ # Metadata queries don't have execution summary
exec_summaries = [result[EXEC_SUMMARY] for result in results[RESULT_LIST]]
+
+ if options.hive_results or exec_summaries[0] is None:
+ return ""
+
combined_summary = CombinedExecSummaries(exec_summaries)
if ref_results is None:
ref_exec_summaries = None
ref_combined_summary = None
else:
- ref_exec_summaries = [result[EXEC_SUMMARY] for result in ref_results[RESULT_LIST]]
+ ref_exec_summaries = [result[EXEC_SUMMARY]
+ for result in ref_results[RESULT_LIST]]
ref_combined_summary = CombinedExecSummaries(ref_exec_summaries)
comparison = ExecSummaryComparison(
@@ -1060,6 +1055,7 @@ def build_exec_summary_str(results, ref_results, for_variability = False):
return str(comparison) + '\n'
+
def build_summary_header(current_impala_version, ref_impala_version):
summary = "Report Generated on {0}\n".format(date.today())
if options.report_description:
@@ -1073,41 +1069,6 @@ def build_summary_header(current_impala_version, ref_impala_version):
summary += 'Baseline Impala Version: {0}\n'.format(ref_impala_version)
return summary
-def write_results_to_datastore(grouped):
- """ Saves results to a database """
-
- from perf_result_datastore import PerfResultDataStore
-
- LOG.info('Saving perf results to database')
- run_date = str(datetime.now())
-
- with PerfResultDataStore(
- host=options.db_host,
- port=options.db_port,
- database_name=options.db_name,
- use_secure_connection=options.use_secure_connection) as data_store:
-
- for results in all_query_results(grouped):
- for query_result in results[RESULT_LIST]:
-
- data_store.insert_execution_result(
- query_name=query_result[QUERY][NAME],
- query_string=query_result[QUERY][QUERY_STR],
- workload_name=query_result[QUERY][WORKLOAD_NAME],
- scale_factor=query_result[QUERY][SCALE_FACTOR],
- file_format=query_result[QUERY][TEST_VECTOR][FILE_FORMAT],
- compression_codec=query_result[QUERY][TEST_VECTOR][COMPRESSION_CODEC],
- compression_type=query_result[QUERY][TEST_VECTOR][COMPRESSION_TYPE],
- num_clients=results[NUM_CLIENTS],
- num_iterations=results[ITERATIONS],
- cluster_name=options.cluster_name,
- executor_name=query_result[EXECUTOR_NAME],
- exec_time=query_result[TIME_TAKEN],
- run_date=run_date,
- version=options.build_version,
- run_info=options.lab_run_info,
- user_name=options.run_user_name,
- runtime_profile = 'N/A' if options.hive_results else query_result[RUNTIME_PROFILE])
if __name__ == "__main__":
"""Workflow:
@@ -1132,7 +1093,6 @@ if __name__ == "__main__":
LOG.error('Could not read reference result file: {0}'.format(e))
ref_grouped = None
- if options.save_to_db: write_results_to_datastore(grouped)
report = Report(grouped, ref_grouped)
ref_impala_version = 'N/A'
[5/9] impala git commit: IMPALA-7477: Batch-oriented query set
construction
Posted by kw...@apache.org.
IMPALA-7477: Batch-oriented query set construction
Rework the row-by-row construction of query result sets in PlanRootSink
so that it materialises an output column at a time. Make some minor
optimisations like preallocating output vectors and initialising
strings more efficiently.
My intent is both to make this faster and to make the QueryResultSet
interface better before IMPALA-4268 does a bunch of surgery on this
part of the code.
Testing:
Ran core tests.
Perf:
Downloaded tpch_parquet.orders via JDBC driver.
Before: 3.01s, After: 2.57s.
Downloaded l_orderkey from tpch_parquet.lineitem.
Before: 1.21s, After: 1.08s.
Change-Id: Ibc87a84c34935d0d5841c7f5528eb802527fa809
Reviewed-on: http://gerrit.cloudera.org:8080/11297
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/impala/commit/b288a6af
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/b288a6af
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/b288a6af
Branch: refs/heads/master
Commit: b288a6af2eda9631b2bad91896ae4bfd2a3fdf30
Parents: 357c0a9
Author: Tim Armstrong <ta...@cloudera.com>
Authored: Sun Aug 19 00:48:47 2018 -0700
Committer: Impala Public Jenkins <im...@cloudera.com>
Committed: Tue Sep 4 17:11:47 2018 +0000
----------------------------------------------------------------------
be/src/exec/plan-root-sink.cc | 26 +--
be/src/exec/plan-root-sink.h | 4 -
be/src/service/hs2-util.cc | 307 +++++++++++++++++++++++---------
be/src/service/hs2-util.h | 15 +-
be/src/service/query-result-set.cc | 115 +++++++-----
be/src/service/query-result-set.h | 14 +-
6 files changed, 320 insertions(+), 161 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/b288a6af/be/src/exec/plan-root-sink.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/plan-root-sink.cc b/be/src/exec/plan-root-sink.cc
index a64dbb9..1f5b2e5 100644
--- a/be/src/exec/plan-root-sink.cc
+++ b/be/src/exec/plan-root-sink.cc
@@ -79,22 +79,11 @@ Status PlanRootSink::Send(RuntimeState* state, RowBatch* batch) {
// Otherwise the consumer is ready. Fill out the rows.
DCHECK(results_ != nullptr);
- // List of expr values to hold evaluated rows from the query
- vector<void*> result_row;
- result_row.resize(output_exprs_.size());
-
- // List of scales for floating point values in result_row
- vector<int> scales;
- scales.resize(result_row.size());
-
int num_to_fetch = batch->num_rows() - current_batch_row;
if (num_rows_requested_ > 0) num_to_fetch = min(num_to_fetch, num_rows_requested_);
- for (int i = 0; i < num_to_fetch; ++i) {
- TupleRow* row = batch->GetRow(current_batch_row);
- GetRowValue(row, &result_row, &scales);
- RETURN_IF_ERROR(results_->AddOneRow(result_row, scales));
- ++current_batch_row;
- }
+ RETURN_IF_ERROR(
+ results_->AddRows(output_expr_evals_, batch, current_batch_row, num_to_fetch));
+ current_batch_row += num_to_fetch;
// Prevent expr result allocations from accumulating.
expr_results_pool_->Clear();
// Signal the consumer.
@@ -146,13 +135,4 @@ Status PlanRootSink::GetNext(
*eos = sender_state_ == SenderState::EOS;
return state->GetQueryStatus();
}
-
-void PlanRootSink::GetRowValue(
- TupleRow* row, vector<void*>* result, vector<int>* scales) {
- DCHECK_GE(result->size(), output_expr_evals_.size());
- for (int i = 0; i < output_expr_evals_.size(); ++i) {
- (*result)[i] = output_expr_evals_[i]->GetValue(row);
- (*scales)[i] = output_expr_evals_[i]->output_scale();
- }
-}
}
http://git-wip-us.apache.org/repos/asf/impala/blob/b288a6af/be/src/exec/plan-root-sink.h
----------------------------------------------------------------------
diff --git a/be/src/exec/plan-root-sink.h b/be/src/exec/plan-root-sink.h
index 1d64b21..300c993 100644
--- a/be/src/exec/plan-root-sink.h
+++ b/be/src/exec/plan-root-sink.h
@@ -118,10 +118,6 @@ class PlanRootSink : public DataSink {
/// Set by GetNext() to indicate to Send() how many rows it should write to results_.
int num_rows_requested_ = 0;
-
- /// Writes a single row into 'result' and 'scales' by evaluating
- /// output_expr_evals_ over 'row'.
- void GetRowValue(TupleRow* row, std::vector<void*>* result, std::vector<int>* scales);
};
}
http://git-wip-us.apache.org/repos/asf/impala/blob/b288a6af/be/src/service/hs2-util.cc
----------------------------------------------------------------------
diff --git a/be/src/service/hs2-util.cc b/be/src/service/hs2-util.cc
index b856556..cf571fb 100644
--- a/be/src/service/hs2-util.cc
+++ b/be/src/service/hs2-util.cc
@@ -18,9 +18,12 @@
#include "service/hs2-util.h"
#include "common/logging.h"
+#include "exprs/scalar-expr-evaluator.h"
#include "runtime/decimal-value.inline.h"
#include "runtime/raw-value.inline.h"
+#include "runtime/row-batch.h"
#include "runtime/types.h"
+#include "util/bit-util.h"
#include <gutil/strings/substitute.h>
@@ -49,7 +52,9 @@ inline bool GetNullBit(const string& nulls, uint32_t row_idx) {
void impala::StitchNulls(uint32_t num_rows_before, uint32_t num_rows_added,
uint32_t start_idx, const string& from, string* to) {
- to->reserve((num_rows_before + num_rows_added + 7) / 8);
+ // Round up to power-of-two to avoid accidentally quadratic behaviour from repeated
+ // small increases in size.
+ to->reserve(BitUtil::RoundUpToPowerOfTwo((num_rows_before + num_rows_added + 7) / 8));
// TODO: This is very inefficient, since we could conceivably go one byte at a time
// (although the operands should stay live in registers in the loop). However doing this
@@ -118,106 +123,246 @@ void impala::TColumnValueToHS2TColumn(const TColumnValue& col_val,
SetNullBit(row_idx, is_null, nulls);
}
+// Specialised per-type implementations of ExprValuesToHS2TColumn.
+
+// Helper to reserve space in hs2Vals->values and hs2Vals->nulls for the values that the
+// different implementations of ExprValuesToHS2TColumn will write.
+template <typename T>
+void ReserveSpace(int start_idx, int num_rows, uint32_t output_row_idx, T* hs2Vals) {
+ int64_t num_output_rows = output_row_idx + num_rows - start_idx;
+ int64_t num_null_bytes = BitUtil::RoundUpNumBytes(num_output_rows);
+ // Round up reserve() arguments to power-of-two to avoid accidentally quadratic
+ // behaviour from repeated small increases in size.
+ hs2Vals->values.reserve(BitUtil::RoundUpToPowerOfTwo(num_output_rows));
+ hs2Vals->nulls.reserve(BitUtil::RoundUpToPowerOfTwo(num_null_bytes));
+}
+
+// Implementation for BOOL.
+static void BoolExprValuesToHS2TColumn(ScalarExprEvaluator* expr_eval, RowBatch* batch,
+ int start_idx, int num_rows, uint32_t output_row_idx,
+ apache::hive::service::cli::thrift::TColumn* column) {
+ ReserveSpace(start_idx, num_rows, output_row_idx, &column->boolVal);
+ FOREACH_ROW_LIMIT(batch, start_idx, num_rows, it) {
+ BooleanVal val = expr_eval->GetBooleanVal(it.Get());
+ column->boolVal.values.push_back(val.val);
+ SetNullBit(output_row_idx, val.is_null, &column->boolVal.nulls);
+ }
+}
+
+// Implementation for TINYINT.
+static void TinyIntExprValuesToHS2TColumn(ScalarExprEvaluator* expr_eval, RowBatch* batch,
+ int start_idx, int num_rows, uint32_t output_row_idx,
+ apache::hive::service::cli::thrift::TColumn* column) {
+ ReserveSpace(start_idx, num_rows, output_row_idx, &column->byteVal);
+ FOREACH_ROW_LIMIT(batch, start_idx, num_rows, it) {
+ TinyIntVal val = expr_eval->GetTinyIntVal(it.Get());
+ column->byteVal.values.push_back(val.val);
+ SetNullBit(output_row_idx, val.is_null, &column->byteVal.nulls);
+ }
+}
+
+// Implementation for SMALLINT.
+static void SmallIntExprValuesToHS2TColumn(ScalarExprEvaluator* expr_eval,
+ RowBatch* batch, int start_idx, int num_rows, uint32_t output_row_idx,
+ apache::hive::service::cli::thrift::TColumn* column) {
+ ReserveSpace(start_idx, num_rows, output_row_idx, &column->i16Val);
+ FOREACH_ROW_LIMIT(batch, start_idx, num_rows, it) {
+ SmallIntVal val = expr_eval->GetSmallIntVal(it.Get());
+ column->i16Val.values.push_back(val.val);
+ SetNullBit(output_row_idx, val.is_null, &column->i16Val.nulls);
+ }
+}
+
+// Implementation for INT.
+static void IntExprValuesToHS2TColumn(ScalarExprEvaluator* expr_eval, RowBatch* batch,
+ int start_idx, int num_rows, uint32_t output_row_idx,
+ apache::hive::service::cli::thrift::TColumn* column) {
+ ReserveSpace(start_idx, num_rows, output_row_idx, &column->i32Val);
+ FOREACH_ROW_LIMIT(batch, start_idx, num_rows, it) {
+ IntVal val = expr_eval->GetIntVal(it.Get());
+ column->i32Val.values.push_back(val.val);
+ SetNullBit(output_row_idx, val.is_null, &column->i32Val.nulls);
+ }
+}
+
+// Implementation for BIGINT.
+static void BigIntExprValuesToHS2TColumn(ScalarExprEvaluator* expr_eval, RowBatch* batch,
+ int start_idx, int num_rows, uint32_t output_row_idx,
+ apache::hive::service::cli::thrift::TColumn* column) {
+ ReserveSpace(start_idx, num_rows, output_row_idx, &column->i64Val);
+ FOREACH_ROW_LIMIT(batch, start_idx, num_rows, it) {
+ BigIntVal val = expr_eval->GetBigIntVal(it.Get());
+ column->i64Val.values.push_back(val.val);
+ SetNullBit(output_row_idx, val.is_null, &column->i64Val.nulls);
+ }
+}
+
+// Implementation for FLOAT.
+static void FloatExprValuesToHS2TColumn(ScalarExprEvaluator* expr_eval, RowBatch* batch,
+ int start_idx, int num_rows, uint32_t output_row_idx,
+ apache::hive::service::cli::thrift::TColumn* column) {
+ ReserveSpace(start_idx, num_rows, output_row_idx, &column->doubleVal);
+ FOREACH_ROW_LIMIT(batch, start_idx, num_rows, it) {
+ FloatVal val = expr_eval->GetFloatVal(it.Get());
+ column->doubleVal.values.push_back(val.val);
+ SetNullBit(output_row_idx, val.is_null, &column->doubleVal.nulls);
+ }
+}
+
+// Implementation for DOUBLE.
+static void DoubleExprValuesToHS2TColumn(ScalarExprEvaluator* expr_eval, RowBatch* batch,
+ int start_idx, int num_rows, uint32_t output_row_idx,
+ apache::hive::service::cli::thrift::TColumn* column) {
+ ReserveSpace(start_idx, num_rows, output_row_idx, &column->doubleVal);
+ FOREACH_ROW_LIMIT(batch, start_idx, num_rows, it) {
+ DoubleVal val = expr_eval->GetDoubleVal(it.Get());
+ column->doubleVal.values.push_back(val.val);
+ SetNullBit(output_row_idx, val.is_null, &column->doubleVal.nulls);
+ }
+}
+
+// Implementation for TIMESTAMP.
+static void TimestampExprValuesToHS2TColumn(ScalarExprEvaluator* expr_eval,
+ RowBatch* batch, int start_idx, int num_rows, uint32_t output_row_idx,
+ apache::hive::service::cli::thrift::TColumn* column) {
+ ReserveSpace(start_idx, num_rows, output_row_idx, &column->stringVal);
+ FOREACH_ROW_LIMIT(batch, start_idx, num_rows, it) {
+ TimestampVal val = expr_eval->GetTimestampVal(it.Get());
+ column->stringVal.values.emplace_back();
+ if (!val.is_null) {
+ TimestampValue value = TimestampValue::FromTimestampVal(val);
+ RawValue::PrintValue(
+ &value, TYPE_TIMESTAMP, -1, &(column->stringVal.values.back()));
+ }
+ SetNullBit(output_row_idx, val.is_null, &column->stringVal.nulls);
+ }
+}
+
+// Implementation for STRING and VARCHAR.
+static void StringExprValuesToHS2TColumn(ScalarExprEvaluator* expr_eval, RowBatch* batch,
+ int start_idx, int num_rows, uint32_t output_row_idx,
+ apache::hive::service::cli::thrift::TColumn* column) {
+ ReserveSpace(start_idx, num_rows, output_row_idx, &column->stringVal);
+ FOREACH_ROW_LIMIT(batch, start_idx, num_rows, it) {
+ StringVal val = expr_eval->GetStringVal(it.Get());
+ if (val.is_null) {
+ column->stringVal.values.emplace_back();
+ } else {
+ column->stringVal.values.emplace_back(reinterpret_cast<char*>(val.ptr), val.len);
+ }
+ SetNullBit(output_row_idx, val.is_null, &column->stringVal.nulls);
+ }
+}
+
+// Implementation for CHAR.
+static void CharExprValuesToHS2TColumn(ScalarExprEvaluator* expr_eval,
+ const TColumnType& type, RowBatch* batch, int start_idx, int num_rows,
+ uint32_t output_row_idx, apache::hive::service::cli::thrift::TColumn* column) {
+ ReserveSpace(start_idx, num_rows, output_row_idx, &column->stringVal);
+ ColumnType char_type = ColumnType::CreateCharType(type.types[0].scalar_type.len);
+ FOREACH_ROW_LIMIT(batch, start_idx, num_rows, it) {
+ StringVal val = expr_eval->GetStringVal(it.Get());
+ if (val.is_null) {
+ column->stringVal.values.emplace_back();
+ } else {
+ column->stringVal.values.emplace_back(
+ reinterpret_cast<const char*>(val.ptr), char_type.len);
+ }
+ SetNullBit(output_row_idx, val.is_null, &column->stringVal.nulls);
+ }
+}
+
+static void DecimalExprValuesToHS2TColumn(ScalarExprEvaluator* expr_eval,
+ const TColumnType& type, RowBatch* batch, int start_idx, int num_rows,
+ uint32_t output_row_idx, apache::hive::service::cli::thrift::TColumn* column) {
+ ReserveSpace(start_idx, num_rows, output_row_idx, &column->stringVal);
+ FOREACH_ROW_LIMIT(batch, start_idx, num_rows, it) {
+ DecimalVal val = expr_eval->GetDecimalVal(it.Get());
+ const ColumnType& decimalType = ColumnType::FromThrift(type);
+ if (val.is_null) {
+ column->stringVal.values.emplace_back();
+ } else {
+ switch (decimalType.GetByteSize()) {
+ case 4:
+ column->stringVal.values.emplace_back(
+ Decimal4Value(val.val4).ToString(decimalType));
+ break;
+ case 8:
+ column->stringVal.values.emplace_back(
+ Decimal8Value(val.val8).ToString(decimalType));
+ break;
+ case 16:
+ column->stringVal.values.emplace_back(
+ Decimal16Value(val.val16).ToString(decimalType));
+ break;
+ default:
+ DCHECK(false) << "bad type: " << decimalType;
+ }
+ }
+ SetNullBit(output_row_idx, val.is_null, &column->stringVal.nulls);
+ }
+}
+
// For V6 and above
-void impala::ExprValueToHS2TColumn(const void* value, const TColumnType& type,
- uint32_t row_idx, thrift::TColumn* column) {
- string* nulls;
+void impala::ExprValuesToHS2TColumn(ScalarExprEvaluator* expr_eval,
+ const TColumnType& type, RowBatch* batch, int start_idx, int num_rows,
+ uint32_t output_row_idx, apache::hive::service::cli::thrift::TColumn* column) {
+ // Dispatch to a templated function for the loop over rows. This avoids branching on
+ // the type for every row.
+ // TODO: instead of relying on stamped out implementations, we could codegen this loop
+ // to inline the expression evaluation into the loop body.
switch (type.types[0].scalar_type.type) {
case TPrimitiveType::NULL_TYPE:
case TPrimitiveType::BOOLEAN:
- column->boolVal.values.push_back(
- value == NULL ? false : *reinterpret_cast<const bool*>(value));
- nulls = &column->boolVal.nulls;
- break;
+ BoolExprValuesToHS2TColumn(
+ expr_eval, batch, start_idx, num_rows, output_row_idx, column);
+ return;
case TPrimitiveType::TINYINT:
- column->byteVal.values.push_back(
- value == NULL ? 0 : *reinterpret_cast<const int8_t*>(value));
- nulls = &column->byteVal.nulls;
- break;
+ TinyIntExprValuesToHS2TColumn(
+ expr_eval, batch, start_idx, num_rows, output_row_idx, column);
+ return;
case TPrimitiveType::SMALLINT:
- column->i16Val.values.push_back(
- value == NULL ? 0 : *reinterpret_cast<const int16_t*>(value));
- nulls = &column->i16Val.nulls;
- break;
+ SmallIntExprValuesToHS2TColumn(
+ expr_eval, batch, start_idx, num_rows, output_row_idx, column);
+ return;
case TPrimitiveType::INT:
- column->i32Val.values.push_back(
- value == NULL ? 0 : *reinterpret_cast<const int32_t*>(value));
- nulls = &column->i32Val.nulls;
- break;
+ IntExprValuesToHS2TColumn(
+ expr_eval, batch, start_idx, num_rows, output_row_idx, column);
+ return;
case TPrimitiveType::BIGINT:
- column->i64Val.values.push_back(
- value == NULL ? 0 : *reinterpret_cast<const int64_t*>(value));
- nulls = &column->i64Val.nulls;
- break;
+ BigIntExprValuesToHS2TColumn(
+ expr_eval, batch, start_idx, num_rows, output_row_idx, column);
+ return;
case TPrimitiveType::FLOAT:
- column->doubleVal.values.push_back(
- value == NULL ? 0.f : *reinterpret_cast<const float*>(value));
- nulls = &column->doubleVal.nulls;
- break;
+ FloatExprValuesToHS2TColumn(
+ expr_eval, batch, start_idx, num_rows, output_row_idx, column);
+ return;
case TPrimitiveType::DOUBLE:
- column->doubleVal.values.push_back(
- value == NULL ? 0.0 : *reinterpret_cast<const double*>(value));
- nulls = &column->doubleVal.nulls;
- break;
+ DoubleExprValuesToHS2TColumn(
+ expr_eval, batch, start_idx, num_rows, output_row_idx, column);
+ return;
case TPrimitiveType::TIMESTAMP:
- column->stringVal.values.push_back("");
- if (value != NULL) {
- RawValue::PrintValue(value, TYPE_TIMESTAMP, -1,
- &(column->stringVal.values.back()));
- }
- nulls = &column->stringVal.nulls;
- break;
+ TimestampExprValuesToHS2TColumn(
+ expr_eval, batch, start_idx, num_rows, output_row_idx, column);
+ return;
case TPrimitiveType::STRING:
case TPrimitiveType::VARCHAR:
- column->stringVal.values.push_back("");
- if (value != NULL) {
- const StringValue* str_val = reinterpret_cast<const StringValue*>(value);
- column->stringVal.values.back().assign(
- static_cast<char*>(str_val->ptr), str_val->len);
- }
- nulls = &column->stringVal.nulls;
- break;
+ StringExprValuesToHS2TColumn(
+ expr_eval, batch, start_idx, num_rows, output_row_idx, column);
+ return;
case TPrimitiveType::CHAR:
- column->stringVal.values.push_back("");
- if (value != NULL) {
- ColumnType char_type = ColumnType::CreateCharType(type.types[0].scalar_type.len);
- column->stringVal.values.back().assign(
- reinterpret_cast<const char*>(value), char_type.len);
- }
- nulls = &column->stringVal.nulls;
- break;
+ CharExprValuesToHS2TColumn(
+ expr_eval, type, batch, start_idx, num_rows, output_row_idx, column);
+ return;
case TPrimitiveType::DECIMAL: {
- // HiveServer2 requires decimal to be presented as string.
- column->stringVal.values.push_back("");
- const ColumnType& decimalType = ColumnType::FromThrift(type);
- if (value != NULL) {
- switch (decimalType.GetByteSize()) {
- case 4:
- column->stringVal.values.back() =
- reinterpret_cast<const Decimal4Value*>(value)->ToString(decimalType);
- break;
- case 8:
- column->stringVal.values.back() =
- reinterpret_cast<const Decimal8Value*>(value)->ToString(decimalType);
- break;
- case 16:
- column->stringVal.values.back() =
- reinterpret_cast<const Decimal16Value*>(value)->ToString(decimalType);
- break;
- default:
- DCHECK(false) << "bad type: " << decimalType;
- }
- }
- nulls = &column->stringVal.nulls;
- break;
+ DecimalExprValuesToHS2TColumn(
+ expr_eval, type, batch, start_idx, num_rows, output_row_idx, column);
+ return;
}
default:
DCHECK(false) << "Unhandled type: "
<< TypeToString(ThriftToType(type.types[0].scalar_type.type));
- return;
}
-
- SetNullBit(row_idx, (value == NULL), nulls);
}
// For V1 -> V5
http://git-wip-us.apache.org/repos/asf/impala/blob/b288a6af/be/src/service/hs2-util.h
----------------------------------------------------------------------
diff --git a/be/src/service/hs2-util.h b/be/src/service/hs2-util.h
index 44ceba6..4f0f973 100644
--- a/be/src/service/hs2-util.h
+++ b/be/src/service/hs2-util.h
@@ -20,16 +20,23 @@
namespace impala {
-/// Utility methods for converting from Impala (either an Expr result or a TColumnValue) to
-/// Hive types (either a thrift::TColumnValue (V1->V5) or a TColumn (V6->).
+class RowBatch;
+class ScalarExprEvaluator;
+
+/// Utility methods for converting from Impala (either an Expr result or a TColumnValue)
+/// to Hive types (either a thrift::TColumnValue (V1->V5) or a TColumn (V6->).
/// For V6->
void TColumnValueToHS2TColumn(const TColumnValue& col_val, const TColumnType& type,
uint32_t row_idx, apache::hive::service::cli::thrift::TColumn* column);
+/// Evaluate 'expr_eval' over the row [start_idx, start_idx + num_rows) from 'batch' into
+/// 'column' with 'type' starting at output_row_idx. The caller is responsible for
+/// calling RuntimeState::GetQueryStatus() to check for expression evaluation errors.
/// For V6->
-void ExprValueToHS2TColumn(const void* value, const TColumnType& type,
- uint32_t row_idx, apache::hive::service::cli::thrift::TColumn* column);
+void ExprValuesToHS2TColumn(ScalarExprEvaluator* expr_eval, const TColumnType& type,
+ RowBatch* batch, int start_idx, int num_rows, uint32_t output_row_idx,
+ apache::hive::service::cli::thrift::TColumn* column);
/// For V1->V5
void TColumnValueToHS2TColumnValue(const TColumnValue& col_val, const TColumnType& type,
http://git-wip-us.apache.org/repos/asf/impala/blob/b288a6af/be/src/service/query-result-set.cc
----------------------------------------------------------------------
diff --git a/be/src/service/query-result-set.cc b/be/src/service/query-result-set.cc
index aacd849..f254176 100644
--- a/be/src/service/query-result-set.cc
+++ b/be/src/service/query-result-set.cc
@@ -20,10 +20,13 @@
#include <sstream>
#include <boost/scoped_ptr.hpp>
+#include "exprs/scalar-expr-evaluator.h"
#include "rpc/thrift-util.h"
#include "runtime/raw-value.h"
+#include "runtime/row-batch.h"
#include "runtime/types.h"
#include "service/hs2-util.h"
+#include "util/bit-util.h"
#include "common/names.h"
@@ -51,18 +54,19 @@ class AsciiQueryResultSet : public QueryResultSet {
virtual ~AsciiQueryResultSet() {}
- /// Convert one row's expr values stored in 'col_values' to ASCII using "\t" as column
+ /// Evaluate 'expr_evals' over rows in 'batch', convert to ASCII using "\t" as column
/// delimiter and store it in this result set.
/// TODO: Handle complex types.
- virtual Status AddOneRow(const vector<void*>& col_values, const vector<int>& scales);
+ virtual Status AddRows(const vector<ScalarExprEvaluator*>& expr_evals, RowBatch* batch,
+ int start_idx, int num_rows) override;
/// Convert TResultRow to ASCII using "\t" as column delimiter and store it in this
/// result set.
- virtual Status AddOneRow(const TResultRow& row);
+ virtual Status AddOneRow(const TResultRow& row) override;
- virtual int AddRows(const QueryResultSet* other, int start_idx, int num_rows);
- virtual int64_t ByteSize(int start_idx, int num_rows);
- virtual size_t size() { return result_set_->size(); }
+ virtual int AddRows(const QueryResultSet* other, int start_idx, int num_rows) override;
+ virtual int64_t ByteSize(int start_idx, int num_rows) override;
+ virtual size_t size() override { return result_set_->size(); }
private:
/// Metadata of the result set
@@ -80,18 +84,20 @@ class HS2ColumnarResultSet : public QueryResultSet {
virtual ~HS2ColumnarResultSet() {}
- /// Add a row of expr values
- virtual Status AddOneRow(const vector<void*>& col_values, const vector<int>& scales);
+ /// Evaluate 'expr_evals' over rows in 'batch' and convert to the HS2 columnar
+ /// representation.
+ virtual Status AddRows(const vector<ScalarExprEvaluator*>& expr_evals, RowBatch* batch,
+ int start_idx, int num_rows) override;
/// Add a row from a TResultRow
- virtual Status AddOneRow(const TResultRow& row);
+ virtual Status AddOneRow(const TResultRow& row) override;
/// Copy all columns starting at 'start_idx' and proceeding for a maximum of 'num_rows'
/// from 'other' into this result set
- virtual int AddRows(const QueryResultSet* other, int start_idx, int num_rows);
+ virtual int AddRows(const QueryResultSet* other, int start_idx, int num_rows) override;
- virtual int64_t ByteSize(int start_idx, int num_rows);
- virtual size_t size() { return num_rows_; }
+ virtual int64_t ByteSize(int start_idx, int num_rows) override;
+ virtual size_t size() override { return num_rows_; }
private:
/// Metadata of the result set
@@ -119,15 +125,17 @@ class HS2RowOrientedResultSet : public QueryResultSet {
virtual ~HS2RowOrientedResultSet() {}
- /// Convert expr values to HS2 TRow and store it in a TRowSet.
- virtual Status AddOneRow(const vector<void*>& col_values, const vector<int>& scales);
+ /// Evaluate 'expr_evals' over rows in 'batch' and convert to the HS2 row-oriented
+ /// representation of TRows stored in a TRowSet.
+ virtual Status AddRows(const vector<ScalarExprEvaluator*>& expr_evals, RowBatch* batch,
+ int start_idx, int num_rows) override;
/// Convert TResultRow to HS2 TRow and store it in a TRowSet
- virtual Status AddOneRow(const TResultRow& row);
+ virtual Status AddOneRow(const TResultRow& row) override;
- virtual int AddRows(const QueryResultSet* other, int start_idx, int num_rows);
- virtual int64_t ByteSize(int start_idx, int num_rows);
- virtual size_t size() { return result_set_->rows.size(); }
+ virtual int AddRows(const QueryResultSet* other, int start_idx, int num_rows) override;
+ virtual int64_t ByteSize(int start_idx, int num_rows) override;
+ virtual size_t size() override { return result_set_->rows.size(); }
private:
/// Metadata of the result set
@@ -158,20 +166,34 @@ QueryResultSet* QueryResultSet::CreateHS2ResultSet(
//////////////////////////////////////////////////////////////////////////////////////////
-Status AsciiQueryResultSet::AddOneRow(
- const vector<void*>& col_values, const vector<int>& scales) {
- int num_col = col_values.size();
+Status AsciiQueryResultSet::AddRows(const vector<ScalarExprEvaluator*>& expr_evals,
+ RowBatch* batch, int start_idx, int num_rows) {
+ DCHECK_GE(batch->num_rows(), start_idx + num_rows);
+ int num_col = expr_evals.size();
DCHECK_EQ(num_col, metadata_.columns.size());
+ vector<int> scales;
+ scales.reserve(num_col);
+ for (ScalarExprEvaluator* expr_eval : expr_evals) {
+ scales.push_back(expr_eval->output_scale());
+ }
+ // Round up to power-of-two to avoid accidentally quadratic behaviour from repeated
+ // small increases in size.
+ result_set_->reserve(
+ BitUtil::RoundUpToPowerOfTwo(result_set_->size() + num_rows - start_idx));
stringstream out_stream;
out_stream.precision(ASCII_PRECISION);
- for (int i = 0; i < num_col; ++i) {
- // ODBC-187 - ODBC can only take "\t" as the delimiter
- out_stream << (i > 0 ? "\t" : "");
- DCHECK_EQ(1, metadata_.columns[i].columnType.types.size());
- RawValue::PrintValue(col_values[i],
- ColumnType::FromThrift(metadata_.columns[i].columnType), scales[i], &out_stream);
+ FOREACH_ROW_LIMIT(batch, start_idx, num_rows, it) {
+ for (int i = 0; i < num_col; ++i) {
+ // ODBC-187 - ODBC can only take "\t" as the delimiter
+ out_stream << (i > 0 ? "\t" : "");
+ DCHECK_EQ(1, metadata_.columns[i].columnType.types.size());
+ RawValue::PrintValue(expr_evals[i]->GetValue(it.Get()),
+ ColumnType::FromThrift(metadata_.columns[i].columnType), scales[i],
+ &out_stream);
+ }
+ result_set_->push_back(out_stream.str());
+ out_stream.str("");
}
- result_set_->push_back(out_stream.str());
return Status::OK();
}
@@ -263,16 +285,18 @@ HS2ColumnarResultSet::HS2ColumnarResultSet(
InitColumns();
}
-// Add a row of expr values
-Status HS2ColumnarResultSet::AddOneRow(
- const vector<void*>& col_values, const vector<int>& scales) {
- int num_col = col_values.size();
+Status HS2ColumnarResultSet::AddRows(const vector<ScalarExprEvaluator*>& expr_evals,
+ RowBatch* batch, int start_idx, int num_rows) {
+ DCHECK_GE(batch->num_rows(), start_idx + num_rows);
+ int num_col = expr_evals.size();
DCHECK_EQ(num_col, metadata_.columns.size());
for (int i = 0; i < num_col; ++i) {
- ExprValueToHS2TColumn(col_values[i], metadata_.columns[i].columnType, num_rows_,
+ const TColumnType& type = metadata_.columns[i].columnType;
+ ScalarExprEvaluator* expr_eval = expr_evals[i];
+ ExprValuesToHS2TColumn(expr_eval, type, batch, start_idx, num_rows, num_rows_,
&(result_set_->columns[i]));
}
- ++num_rows_;
+ num_rows_ += num_rows;
return Status::OK();
}
@@ -427,16 +451,21 @@ HS2RowOrientedResultSet::HS2RowOrientedResultSet(
}
}
-Status HS2RowOrientedResultSet::AddOneRow(
- const vector<void*>& col_values, const vector<int>& scales) {
- int num_col = col_values.size();
+Status HS2RowOrientedResultSet::AddRows(const vector<ScalarExprEvaluator*>& expr_evals,
+ RowBatch* batch, int start_idx, int num_rows) {
+ DCHECK_GE(batch->num_rows(), start_idx + num_rows);
+ int num_col = expr_evals.size();
DCHECK_EQ(num_col, metadata_.columns.size());
- result_set_->rows.push_back(TRow());
- TRow& trow = result_set_->rows.back();
- trow.colVals.resize(num_col);
- for (int i = 0; i < num_col; ++i) {
- ExprValueToHS2TColumnValue(
- col_values[i], metadata_.columns[i].columnType, &(trow.colVals[i]));
+ result_set_->rows.reserve(
+ BitUtil::RoundUpToPowerOfTwo(result_set_->rows.size() + num_rows - start_idx));
+ FOREACH_ROW_LIMIT(batch, start_idx, num_rows, it) {
+ result_set_->rows.push_back(TRow());
+ TRow& trow = result_set_->rows.back();
+ trow.colVals.resize(num_col);
+ for (int i = 0; i < num_col; ++i) {
+ ExprValueToHS2TColumnValue(expr_evals[i]->GetValue(it.Get()),
+ metadata_.columns[i].columnType, &(trow.colVals[i]));
+ }
}
return Status::OK();
}
http://git-wip-us.apache.org/repos/asf/impala/blob/b288a6af/be/src/service/query-result-set.h
----------------------------------------------------------------------
diff --git a/be/src/service/query-result-set.h b/be/src/service/query-result-set.h
index e0c88d7..fa39d73 100644
--- a/be/src/service/query-result-set.h
+++ b/be/src/service/query-result-set.h
@@ -27,6 +27,9 @@
namespace impala {
+class RowBatch;
+class ScalarExprEvaluator;
+
/// Wraps a client-API specific result representation, and implements the logic required
/// to translate into that format from Impala's row format.
///
@@ -36,12 +39,11 @@ class QueryResultSet {
QueryResultSet() {}
virtual ~QueryResultSet() {}
- /// Add a single row to this result set. The row is a vector of pointers to values,
- /// whose memory belongs to the caller. 'scales' contains the scales for decimal values
- /// (# of digits after decimal), with -1 indicating no scale specified or the
- /// corresponding value is not a decimal.
- virtual Status AddOneRow(
- const std::vector<void*>& row, const std::vector<int>& scales) = 0;
+ /// Add 'num_rows' rows to the result set, obtained by evaluating 'expr_evals' over
+ /// the rows in 'batch' starting at start_idx. Batch must contain at least
+ /// ('start_idx' + 'num_rows') rows.
+ virtual Status AddRows(const std::vector<ScalarExprEvaluator*>& expr_evals,
+ RowBatch* batch, int start_idx, int num_rows) = 0;
/// Add the TResultRow to this result set. When a row comes from a DDL/metadata
/// operation, the row in the form of TResultRow.
[8/9] impala git commit: Fix generate_junitxml for python2.6.
Posted by kw...@apache.org.
Fix generate_junitxml for python2.6.
With python 2.6, the syntax "{}".format(1) doesn't work:
$docker run centos:6 python -c 'print "{}".format(1)'
Traceback (most recent call last):
File "<string>", line 1, in <module>
ValueError: zero length field name in format
generate_junitxml was using this incantation and failing.
I've updated the syntax to be py2.6-friendly, and tested
it like so:
$docker run -v $(pwd):/mnt centos:6 bash -c "yum install -y python-argparse; /mnt/lib/python/impala_py_lib/jenkins/generate_junitxml.py --phase phase --step step --stdout out --stderr err; cat /extra_junit_xml_logs/*.xml"
[output from yum...]
Installed:
python-argparse.noarch 0:1.2.1-2.1.el6
Complete!
Generated: ./extra_junit_xml_logs/generate_junitxml.phase.step.20180904_18_04_56.xml
<?xml version="1.0" ?>
<testsuites errors="0" failures="0" tests="1" time="0.0">
<testsuite disabled="0" errors="0" failures="0" file="None" log="None" name="generate_junitxml.phase.step" skipped="0" tests="1" time="0" timestamp="2018-09-04 18:04:56+00:00" url="None">
<testcase classname="generate_junitxml.phase" name="step">
<system-out>
out
</system-out>
<system-err>
err
</system-err>
</testcase>
</testsuite>
</testsuites>
Change-Id: Ic0c1e837a9ed6c2d59906aed1d1098bde6f5d815
Reviewed-on: http://gerrit.cloudera.org:8080/11384
Reviewed-by: David Knupp <dk...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/impala/commit/304e02cf
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/304e02cf
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/304e02cf
Branch: refs/heads/master
Commit: 304e02cf6238a3c1e9537337f802bcf7a92df5db
Parents: 5f5c861
Author: Philip Zeyliger <ph...@cloudera.com>
Authored: Tue Sep 4 11:05:07 2018 -0700
Committer: Impala Public Jenkins <im...@cloudera.com>
Committed: Tue Sep 4 21:58:08 2018 +0000
----------------------------------------------------------------------
.../impala_py_lib/jenkins/generate_junitxml.py | 18 ++++++++++--------
1 file changed, 10 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/304e02cf/lib/python/impala_py_lib/jenkins/generate_junitxml.py
----------------------------------------------------------------------
diff --git a/lib/python/impala_py_lib/jenkins/generate_junitxml.py b/lib/python/impala_py_lib/jenkins/generate_junitxml.py
index 97dcbdd..4a1f22f 100755
--- a/lib/python/impala_py_lib/jenkins/generate_junitxml.py
+++ b/lib/python/impala_py_lib/jenkins/generate_junitxml.py
@@ -97,10 +97,10 @@ class JunitReport(object):
def add_testsuite_element(self):
"""Create the testsuite element."""
self.testsuite_element = ET.SubElement(self.root_element, "testsuite")
+ self.testsuite_element.set("name", "{name}.{phase}.{step}".format(
+ name=SCRIPT_NAME, phase=self.phase, step=self.step))
self.testsuite_element.set(
- "name", "{}.{}.{}".format(SCRIPT_NAME, self.phase, self.step))
- self.testsuite_element.set(
- "timestamp", "{}+00:00".format(self.utc_time.strftime('%Y-%m-%d %H:%M:%S')))
+ "timestamp", "{ts}+00:00".format(ts=self.utc_time.strftime('%Y-%m-%d %H:%M:%S')))
self.testsuite_element.set("disabled", "0")
self.testsuite_element.set("errors", "0")
self.testsuite_element.set("failures", "0")
@@ -114,7 +114,8 @@ class JunitReport(object):
def add_testcase_element(self):
"""Create the testcase element."""
self.testcase_element = ET.SubElement(self.testsuite_element, "testcase")
- self.testcase_element.set("classname", "{}.{}".format(SCRIPT_NAME, self.phase))
+ self.testcase_element.set("classname", "{name}.{phase}".format(
+ name=SCRIPT_NAME, phase=self.phase))
self.testcase_element.set("name", self.step)
def set_error(self):
@@ -133,7 +134,8 @@ class JunitReport(object):
output_type: [string] either out or err
file_or_string: a path to a file containing the content, or a plain string
"""
- output = ET.SubElement(self.testcase_element, "system-{}".format(output_type))
+ output = ET.SubElement(self.testcase_element,
+ "system-{output_type}".format(output_type=output_type))
output.text = JunitReport.get_xml_content(file_or_string)
def to_file(self, junitxml_logdir=JUNITXML_LOGDIR):
@@ -155,9 +157,9 @@ class JunitReport(object):
else:
raise
- filename = '{}.{}.xml'.format(
- self.testsuite_element.attrib['name'],
- self.utc_time.strftime('%Y%m%d_%H_%M_%S')
+ filename = '{name}.{ts}.xml'.format(
+ name=self.testsuite_element.attrib['name'],
+ ts=self.utc_time.strftime('%Y%m%d_%H_%M_%S')
)
junit_log_file = os.path.join(junitxml_logdir, filename)
[2/9] impala git commit: [DOCS] Changed the Dita output format to
XHTML
Posted by kw...@apache.org.
[DOCS] Changed the Dita output format to XHTML
Changed the output format to XHTML to work around a bug that
generates an incorrect ids for sub-concepts. The bug prevented
cross-referencing sections within docs.
Change-Id: I26084fc8a2fe180093c61a105613791a49865732
Reviewed-on: http://gerrit.cloudera.org:8080/11375
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Reviewed-by: Jim Apple <jb...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/impala/commit/1e7d627d
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/1e7d627d
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/1e7d627d
Branch: refs/heads/master
Commit: 1e7d627d0d32687042b23613d3ef8d58334ff97b
Parents: 37c6699
Author: Alex Rodoni <ar...@cloudera.com>
Authored: Fri Aug 31 13:30:30 2018 -0700
Committer: Alex Rodoni <ar...@cloudera.com>
Committed: Sat Sep 1 00:51:15 2018 +0000
----------------------------------------------------------------------
docs/Makefile | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/1e7d627d/docs/Makefile
----------------------------------------------------------------------
diff --git a/docs/Makefile b/docs/Makefile
index fc2f8e9..837a336 100644
--- a/docs/Makefile
+++ b/docs/Makefile
@@ -26,7 +26,7 @@ pdf: build/impala.pdf
ALL_DEPS=Makefile impala.ditamap shared/*.xml images/* topics/*.xml
build/html/index.html: impala_html.ditaval ${ALL_DEPS}
- ./build-doc.sh html5 $(dir $@) $< gen-html.log
+ ./build-doc.sh xhtml $(dir $@) $< gen-html.log
build/impala.pdf: impala_pdf.ditaval ${ALL_DEPS}
./build-doc.sh pdf $(dir $@) $< gen-pdf.log
[4/9] impala git commit: IMPALA-7490: fix uninitialized variables in
load-data.py
Posted by kw...@apache.org.
IMPALA-7490: fix uninitialized variables in load-data.py
Fixes use of an uninitialized variable in bin/load-data.py
I found the following error message in a failed build, which is quite
misleading:
Traceback (most recent call last):
File "bin/load-data.py", line 495, in <module>
if __name__ == "__main__": main()
File "bin/load-data.py", line 459, in main
impala_exec_query_files_parallel(thread_pool, impala_create_files)
File "bin/load-data.py", line 297, in impala_exec_query_files_parallel
exec_query_files_parallel(thread_pool, query_files, 'impala')
File "bin/load-data.py", line 291, in exec_query_files_parallel
for result in thread_pool.imap_unordered(execution_function, query_files):
File "/usr/lib/python2.7/multiprocessing/pool.py", line 659, in next
raise value
UnboundLocalError: local variable 'query' referenced before assignment
The error is trown from the 'execution_function' which is
'exec_impala_query_from_file' in my case. Should not use 'query' if it's
undefined.
Change-Id: If0dd56a9b78a60b3a9f04d9f61e93b4b5d066b76
Reviewed-on: http://gerrit.cloudera.org:8080/11330
Reviewed-by: Quanlong Huang <hu...@gmail.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/impala/commit/357c0a95
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/357c0a95
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/357c0a95
Branch: refs/heads/master
Commit: 357c0a959d1a3a9cb2662773a407acfee3ca6fe2
Parents: fe613e3
Author: stiga-huang <hu...@gmail.com>
Authored: Sat Aug 25 19:09:23 2018 -0700
Committer: Impala Public Jenkins <im...@cloudera.com>
Committed: Sat Sep 1 12:00:43 2018 +0000
----------------------------------------------------------------------
bin/load-data.py | 6 +++++-
1 file changed, 5 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/357c0a95/bin/load-data.py
----------------------------------------------------------------------
diff --git a/bin/load-data.py b/bin/load-data.py
index 7754147..8878372 100755
--- a/bin/load-data.py
+++ b/bin/load-data.py
@@ -195,6 +195,7 @@ def exec_impala_query_from_file(file_name):
is_success = True
impala_client = ImpalaBeeswaxClient(options.impalad, use_kerberos=options.use_kerberos)
output_file = file_name + ".log"
+ query = None
with open(output_file, 'w') as out_file:
try:
impala_client.connect()
@@ -206,7 +207,10 @@ def exec_impala_query_from_file(file_name):
result = impala_client.execute(query)
out_file.write("{0}\n{1}\n".format(query, result))
except Exception as e:
- out_file.write("ERROR: {0}\n".format(query))
+ if query:
+ out_file.write("ERROR: {0}\n".format(query))
+ else:
+ out_file.write("Encounter errors before parsing any queries.\n")
traceback.print_exc(file=out_file)
is_success = False
[3/9] impala git commit: IMPALA-7508: Add Impala Python GDB module
Posted by kw...@apache.org.
IMPALA-7508: Add Impala Python GDB module
This patch adds a new Impala Python GDB module, and implements
a couple of covenience commands to make core dump analysis
easier.
The initial commands let us find queries and fragment instances
currently executing in an impalad at the time the daemon crashed:
(gdb) source impala-gdb.py
(gdb) find-query-ids
f74c863dff66a34d:1d983cc300000000
364525e12495932b:73f5dd0200000000
bc4a3eec25481981:edda04b800000000
(gdb) find-fragment-instances
Fragment Instance Id Thread IDs
364525e12495932b:73f5dd02000000a2 [69]
364525e12495932b:73f5dd0200000171 [196, 136]
bc4a3eec25481981:edda04b8000001a8 [252, 237, 206]
f74c863dff66a34d:1d983cc30000009b [200, 14, 13, 12, 6, 5, 3, 2]
f74c863dff66a34d:1d983cc30000013a [4]
The commands have only been tested with Impala 2.12, and are not
expected to work with older versions since it uses ThreadDebugInfo
stuff from IMPALA-6416.
It is hoped that people contribute more commands to the module.
Change-Id: I24e7026f2265954ed592d6f62110cf8cb2c2202a
Reviewed-on: http://gerrit.cloudera.org:8080/11352
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/impala/commit/fe613e32
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/fe613e32
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/fe613e32
Branch: refs/heads/master
Commit: fe613e3247c43bf507b30fd279b73768e36ff5b5
Parents: 1e7d627
Author: Zoram Thanga <zo...@cloudera.com>
Authored: Wed Aug 29 14:57:55 2018 -0700
Committer: Impala Public Jenkins <im...@cloudera.com>
Committed: Sat Sep 1 03:54:29 2018 +0000
----------------------------------------------------------------------
bin/rat_exclude_files.txt | 1 +
lib/python/impala_py_lib/gdb/README.md | 43 +++++++++++
lib/python/impala_py_lib/gdb/impala-gdb.py | 99 +++++++++++++++++++++++++
3 files changed, 143 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/fe613e32/bin/rat_exclude_files.txt
----------------------------------------------------------------------
diff --git a/bin/rat_exclude_files.txt b/bin/rat_exclude_files.txt
index c0c8fe2..cb5e022 100644
--- a/bin/rat_exclude_files.txt
+++ b/bin/rat_exclude_files.txt
@@ -88,6 +88,7 @@ docs/README.md
docker/README.md
be/src/thirdparty/pcg-cpp-0.98/README.md
lib/python/README.md
+lib/python/impala_py_lib/gdb/README.md
# http://www.apache.org/legal/src-headers.html: "Test data for which the addition of a
# source header would cause the tests to fail."
http://git-wip-us.apache.org/repos/asf/impala/blob/fe613e32/lib/python/impala_py_lib/gdb/README.md
----------------------------------------------------------------------
diff --git a/lib/python/impala_py_lib/gdb/README.md b/lib/python/impala_py_lib/gdb/README.md
new file mode 100644
index 0000000..e9ef855
--- /dev/null
+++ b/lib/python/impala_py_lib/gdb/README.md
@@ -0,0 +1,43 @@
+## What is impala-gdb.py?
+
+This is a collection of Python GDB functions (macros) that can be invoked from a GDB
+session to aid analysis of Impala core dumps.
+
+To use the functions, source impala-gdb.py as follows:
+
+```
+
+(gdb) source $IMPALA_HOME/lib/python/impala_py_lib/gdb/impala-gdb.py
+
+```
+
+Currently, the following functions are available:
+
+```
+1. find-query-ids
+2. find-fragment-instances
+```
+
+Here's a sample interaction:
+
+```
+(gdb) help find-query-ids
+Find IDs of all queries this impalad is currently executing.
+(gdb) find-query-ids
+f74c863dff66a34d:1d983cc300000000
+364525e12495932b:73f5dd0200000000
+bc4a3eec25481981:edda04b800000000
+(gdb)
+(gdb)
+(gdb) help find-fragment-instances
+Find all query fragment instance to thread Id mappings in this impalad.
+(gdb) find-fragment-instances
+Fragment Instance Id Thread IDs
+
+364525e12495932b:73f5dd02000000a2 [69]
+364525e12495932b:73f5dd0200000171 [196, 136]
+bc4a3eec25481981:edda04b8000001a8 [252, 237, 206]
+f74c863dff66a34d:1d983cc30000009b [200, 14, 13, 12, 6, 5, 3, 2]
+f74c863dff66a34d:1d983cc30000013a [4]
+(gdb)
+```
http://git-wip-us.apache.org/repos/asf/impala/blob/fe613e32/lib/python/impala_py_lib/gdb/impala-gdb.py
----------------------------------------------------------------------
diff --git a/lib/python/impala_py_lib/gdb/impala-gdb.py b/lib/python/impala_py_lib/gdb/impala-gdb.py
new file mode 100644
index 0000000..904b771
--- /dev/null
+++ b/lib/python/impala_py_lib/gdb/impala-gdb.py
@@ -0,0 +1,99 @@
+#
+# 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.
+#
+# A collection of useful Python GDB modules and commands for
+# debugging Impala core dumps.
+#
+import gdb
+from collections import defaultdict
+
+
+def get_fragment_instances():
+ # Walk the threadlist to find fragment instance ids. Assumes IMPALA-6416, so
+ # this will not work with releases older than Impala 2.12. It may be possible
+ # to search for FragmentInstanceState::Exec() in the older releases to get
+ # to the FInstIDs. Returns a dictionary of FInstID->[gdb thread ID].
+ # Note that multiple threads might be doing tasks for the same FInstID.
+
+ fragment_instances = defaultdict(list)
+ for thread in gdb.selected_inferior().threads():
+ thread.switch()
+
+ f = gdb.newest_frame()
+ while f:
+ # Skip unresolved frame
+ if gdb.Frame.name(f) is None:
+ f = gdb.Frame.older(f)
+ continue
+ if 'impala::Thread::SuperviseThread' in gdb.Frame.name(f):
+ gdb.Frame.select(f)
+ block = gdb.Frame.block(f)
+ gdb.lookup_symbol('parent_thread_info', block)
+ p = f.read_var('parent_thread_info')
+ # No valid parent_thread_info pointer
+ if not p:
+ break
+ v = p.dereference()
+ fi = str(v['instance_id_'])
+ if ':' in fi:
+ fragment_instances[fi.strip('"')].append(thread.num)
+ break
+ f = gdb.Frame.older(f)
+ return fragment_instances
+
+
+class FindFragmentInstances(gdb.Command):
+ """Find all query fragment instance to thread Id mappings in this impalad."""
+
+ def __init__(self):
+ super(FindFragmentInstances, self).__init__(
+ "find-fragment-instances",
+ gdb.COMMAND_SUPPORT,
+ gdb.COMPLETE_NONE,
+ False)
+
+ def invoke(self, arg, from_tty):
+ fragment_instances = get_fragment_instances()
+ print('Fragment Instance Id\tThread IDs\n')
+ for fid in sorted(fragment_instances):
+ print("{}\t{}".format(fid, fragment_instances[fid]))
+
+
+class FindQueryIds(gdb.Command):
+ """Find IDs of all queries this impalad is currently executing."""
+
+ def __init__(self):
+ super(FindQueryIds, self).__init__(
+ "find-query-ids",
+ gdb.COMMAND_SUPPORT,
+ gdb.COMPLETE_NONE,
+ False)
+
+ def invoke(self, arg, from_tty):
+ fragment_instances = get_fragment_instances()
+ query_ids = set()
+ for fi in fragment_instances:
+ qid_hi, qid_low = fi.split(':')
+ qid_low = format(int(qid_low, 16) & 0xFFFFFFFFFFFF0000, 'x')
+ query_ids.add("{}:{}".format(qid_hi, qid_low))
+ print('\n'.join(query_ids))
+
+
+# Instantiate the command classes so they can be invoked from within GDB
+FindFragmentInstances()
+FindQueryIds()
[6/9] impala git commit: IMPALA-6919: [DOCS] Document the COMMENT ON
statement
Posted by kw...@apache.org.
IMPALA-6919: [DOCS] Document the COMMENT ON statement
Change-Id: Ibb7ce9fa20fb1d6551d718c8ec25c746a72b40d9
Reviewed-on: http://gerrit.cloudera.org:8080/11370
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Reviewed-by: Vuk Ercegovac <ve...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/impala/commit/c8d21d01
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/c8d21d01
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/c8d21d01
Branch: refs/heads/master
Commit: c8d21d0137db9bf4c5cac294f8a25aa76da6fcb6
Parents: b288a6a
Author: Alex Rodoni <ar...@cloudera.com>
Authored: Thu Aug 30 17:40:02 2018 -0700
Committer: Alex Rodoni <ar...@cloudera.com>
Committed: Tue Sep 4 17:44:21 2018 +0000
----------------------------------------------------------------------
docs/impala.ditamap | 1 +
docs/topics/impala_comment.xml | 110 ++++++++++++++++++++++++++++++++++++
2 files changed, 111 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/c8d21d01/docs/impala.ditamap
----------------------------------------------------------------------
diff --git a/docs/impala.ditamap b/docs/impala.ditamap
index 73c5d23..1c7a69c 100644
--- a/docs/impala.ditamap
+++ b/docs/impala.ditamap
@@ -131,6 +131,7 @@ under the License.
<topicref href="topics/impala_dml.xml"/>
<topicref href="topics/impala_alter_table.xml"/>
<topicref href="topics/impala_alter_view.xml"/>
+ <topicref href="topics/impala_comment.xml"/>
<topicref href="topics/impala_compute_stats.xml"/>
<topicref audience="hidden" href="topics/impala_create_data_source.xml"/>
<topicref href="topics/impala_create_database.xml"/>
http://git-wip-us.apache.org/repos/asf/impala/blob/c8d21d01/docs/topics/impala_comment.xml
----------------------------------------------------------------------
diff --git a/docs/topics/impala_comment.xml b/docs/topics/impala_comment.xml
new file mode 100644
index 0000000..00bfc11
--- /dev/null
+++ b/docs/topics/impala_comment.xml
@@ -0,0 +1,110 @@
+<?xml version="1.0" encoding="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.
+-->
+<!DOCTYPE concept PUBLIC "-//OASIS//DTD DITA Concept//EN" "concept.dtd">
+<concept id="comment">
+
+ <title>COMMENT Statement</title>
+
+ <titlealts audience="PDF">
+
+ <navtitle>COMMENT</navtitle>
+
+ </titlealts>
+
+ <prolog>
+ <metadata>
+ <data name="Category" value="Impala"/>
+ <data name="Category" value="SQL"/>
+ <data name="Category" value="DDL"/>
+ <data name="Category" value="Tables"/>
+ <data name="Category" value="Hive"/>
+ <data name="Category" value="Metastore"/>
+ <data name="Category" value="ETL"/>
+ <data name="Category" value="Ingest"/>
+ <data name="Category" value="Developers"/>
+ <data name="Category" value="Data Analysts"/>
+ </metadata>
+ </prolog>
+
+ <conbody>
+
+ <p>
+ The <codeph>COMMENT</codeph> statement adds, changes, or removes a comment about a
+ database, a table, or a column.
+ </p>
+
+ <p>
+ You can alternatively use the <codeph>CREATE</codeph> and <codeph>ALTER</codeph>
+ statements to add comments to the objects.
+ </p>
+
+ <p>
+ You can view the comment on a database, a table, or a column using the
+ <codeph>SHOW</codeph> or <codeph>DESCRIBE</codeph> statement.
+ </p>
+
+ <p conref="../shared/impala_common.xml#common/syntax_blurb"/>
+
+<codeblock>COMMENT ON DATABASE <varname>db_name</varname> IS {'<varname>comment</varname>' | NULL}</codeblock>
+
+<codeblock>COMMENT ON TABLE [<varname>db_name</varname>.]<varname>table_name</varname> IS {'<varname>comment</varname>' | NULL}</codeblock>
+
+<codeblock>COMMENT ON COLUMN [<varname>db_name</varname>.]<varname>table_name</varname>.<varname>column_name</varname> IS {'<varname>comment</varname>' | NULL}</codeblock>
+
+ <p>
+ <b>Parameters:</b>
+ </p>
+
+ <ul>
+ <li>
+ <varname>db_name</varname>: Specify the database name if not for the current database.
+ </li>
+
+ <li>
+ <codeph>NULL</codeph>: If given for the comment, removes the existing comment.
+ </li>
+
+ <li>
+ The <varname>comment</varname> string can be up to 256 characters long.
+ </li>
+ </ul>
+
+ <p>
+ <b>Privileges required:</b>
+ </p>
+
+ <p>
+ To add a comment, the <codeph>ALTER</codeph> privilege on the object is required.
+ </p>
+
+ <p>
+ To view a comment, the <codeph>SELECT</codeph>, <codeph>INSERT</codeph>, or
+ <codeph>REFRESH</codeph> on the object is required.
+ </p>
+
+ <p conref="../shared/impala_common.xml#common/usage_notes_blurb"/>
+
+ <p>
+ <b>Added in:</b> <keyword keyref="impala31"/>
+ </p>
+
+ </conbody>
+
+</concept>
[7/9] impala git commit: [diagnostics] Make --minidump_dir consistent
with Impala's --minidump_path
Posted by kw...@apache.org.
[diagnostics] Make --minidump_dir consistent with Impala's --minidump_path
Currently, the diagnostics script expects a full path to the actual
directory to which process minidumps are written. This is however
incosistent with Impala's configuration --minidump_path.
Impala creates a subdirectory under FLAGS_minidump_path (for ex:
<FLAGS_minidump_path>/impalad) to which it writes the minidumps.
This commit fixes the diagnostic script input --minidump_dir to be
consistent with the above behavior from Impala. It now looks for
minidumps under the directory <--minidump_path>/<process-name>
The users of this script are expected to fix their input args
accordingly.
Change-Id: I9e59f108a1f29a33768a39d0f4554d96e2dcd381
Reviewed-on: http://gerrit.cloudera.org:8080/11353
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Reviewed-by: Bharath Vissapragada <bh...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/impala/commit/5f5c8612
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/5f5c8612
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/5f5c8612
Branch: refs/heads/master
Commit: 5f5c8612f6db39624b8988d2587781e3a3a29d3f
Parents: c8d21d0
Author: Bharath Vissapragada <bh...@cloudera.com>
Authored: Wed Aug 29 15:15:43 2018 -0700
Committer: Bharath Vissapragada <bh...@cloudera.com>
Committed: Tue Sep 4 20:17:37 2018 +0000
----------------------------------------------------------------------
bin/diagnostics/collect_diagnostics.py | 15 ++++++++++-----
1 file changed, 10 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/5f5c8612/bin/diagnostics/collect_diagnostics.py
----------------------------------------------------------------------
diff --git a/bin/diagnostics/collect_diagnostics.py b/bin/diagnostics/collect_diagnostics.py
index 8ef0385..f4b3298 100755
--- a/bin/diagnostics/collect_diagnostics.py
+++ b/bin/diagnostics/collect_diagnostics.py
@@ -71,7 +71,7 @@ from threading import Timer
#
# Collect 5 breakpad minidumps from a statestored process 5s apart.
# python collect_diagnostics.py --pid $(pidof statestored) --minidumps 5 5
-# --minidumps_dir /var/log/statestored/minidumps
+# --minidumps_dir /var/log/impala-minidumps
#
#
class Command(object):
@@ -126,6 +126,9 @@ class ImpalaDiagnosticsHandler(object):
# Name of the Impala process for which diagnostics should be collected.
self.target_process_name = self.get_target_process_name()
+ self.minidump_search_path = os.path.join(self.args.minidumps_dir,
+ self.target_process_name)
+
self.java_home = self.get_java_home_from_env()
if not self.java_home and args.java_home:
self.java_home = os.path.abspath(args.java_home)
@@ -276,7 +279,7 @@ class ImpalaDiagnosticsHandler(object):
fallback_to_minidump = False
if not self.pstack_cmd:
# Fall back to collecting a minidump if pstack is not installed.
- if not os.path.exists(self.args.minidumps_dir):
+ if not os.path.exists(self.minidump_search_path):
logging.info("Skipping pstacks since pstack binary couldn't be located. Provide "
+ "--minidumps_dir for collecting minidumps instead.")
# At this point, we can't proceed since we have nothing to collect.
@@ -348,8 +351,8 @@ class ImpalaDiagnosticsHandler(object):
def copy_minidumps(self, target, start_ts):
"""Copies mindumps with create time >= start_ts to 'target' directory."""
logging.info("Copying minidumps from %s to %s with ctime >= %s"
- % (self.args.minidumps_dir, target, start_ts))
- for filename in glob.glob(os.path.join(self.args.minidumps_dir, "*.dmp")):
+ % (self.minidump_search_path, target, start_ts))
+ for filename in glob.glob(os.path.join(self.minidump_search_path, "*.dmp")):
try:
minidump_ctime = self.get_minidump_create_timestamp(filename)
if minidump_ctime >= math.floor(start_ts):
@@ -526,7 +529,9 @@ def get_args_parser():
help="Collect breakpad minidumps for the Impala process. Requires --minidumps_dir\
be set.")
parser.add_argument("--minidumps_dir", action="store", dest="minidumps_dir", default="",
- help="Path of the directory to which Impala process' minidumps are written")
+ help="Path of the directory to which Impala process' minidumps are written. Looks\
+ for minidumps in this path's subdirectory that is named after the target process\
+ name.")
parser.add_argument("--profiles_dir", action="store", dest="profiles_dir", default="",
help="Path of the profiles directory to be included in the diagnostics output.")
parser.add_argument("--profiles_max_size_limit", action="store",
[9/9] impala git commit: Add bin/jenkins/finalize.sh
Posted by kw...@apache.org.
Add bin/jenkins/finalize.sh
This provides a place for doing cleanup and error checking at the end
of Jenkins builds. Currently, it only checks dmesg for OOMs.
Testing:
- https://jenkins.impala.io/job/ubuntu-16.04-from-scratch/3076/
Change-Id: If4d223b6e441f01a5edad9d6f9ca69c6de023461
Reviewed-on: http://gerrit.cloudera.org:8080/11322
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/impala/commit/3433d8fe
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/3433d8fe
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/3433d8fe
Branch: refs/heads/master
Commit: 3433d8fe1f5a24c87ee80c9d86e79c4ecda87490
Parents: 304e02c
Author: Thomas Tauber-Marshall <tm...@cloudera.com>
Authored: Fri Aug 24 21:01:33 2018 +0000
Committer: Impala Public Jenkins <im...@cloudera.com>
Committed: Wed Sep 5 03:08:05 2018 +0000
----------------------------------------------------------------------
bin/jenkins/all-tests.sh | 9 ++++++++-
bin/jenkins/finalize.sh | 32 ++++++++++++++++++++++++++++++++
2 files changed, 40 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/3433d8fe/bin/jenkins/all-tests.sh
----------------------------------------------------------------------
diff --git a/bin/jenkins/all-tests.sh b/bin/jenkins/all-tests.sh
index 1e73722..d29f383 100644
--- a/bin/jenkins/all-tests.sh
+++ b/bin/jenkins/all-tests.sh
@@ -27,4 +27,11 @@ cd "${IMPALA_HOME}"
export IMPALA_MAVEN_OPTIONS="-U"
source bin/bootstrap_development.sh
-bin/run-all-tests.sh
+
+RET_CODE=0
+if ! bin/run-all-tests.sh; then
+ RET_CODE=1
+fi
+
+bin/jenkins/finalize.sh
+exit $RET_CODE
http://git-wip-us.apache.org/repos/asf/impala/blob/3433d8fe/bin/jenkins/finalize.sh
----------------------------------------------------------------------
diff --git a/bin/jenkins/finalize.sh b/bin/jenkins/finalize.sh
new file mode 100755
index 0000000..ada4019
--- /dev/null
+++ b/bin/jenkins/finalize.sh
@@ -0,0 +1,32 @@
+#!/bin/bash
+# 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.
+
+# Do some error checking and generate junit symptoms after running a build.
+
+set -euo pipefail
+trap 'echo Error in $0 at line $LINENO: $(cd "'$PWD'" && awk "NR == $LINENO" $0)' ERR
+
+rm -rf "${IMPALA_HOME}"/logs_system
+mkdir -p "${IMPALA_HOME}"/logs_system
+dmesg > "${IMPALA_HOME}"/logs_system/dmesg
+
+# Check dmesg for OOMs and generate a symptom if present.
+if [[ $(grep "Out of memory" "${IMPALA_HOME}"/logs_system/dmesg) ]]; then
+ "${IMPALA_HOME}"/bin/generate_junitxml.py --phase finalize --step dmesg \
+ --stdout "${IMPALA_HOME}"/logs_system/dmesg --error "Process was OOM killed."
+fi