| # 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. |
| # |
| # Impala tests for Hive Metastore, covering the expected propagation |
| # of metadata from Hive to Impala or Impala to Hive. Each test |
| # modifies the metadata via Hive and checks that the modification |
| # succeeded by querying Impala, or vice versa. |
| # |
| # TODO: For each test, verify all the metadata available via Hive and |
| # Impala, in all the possible ways of validating that metadata. |
| |
| |
| import pytest |
| import random |
| import string |
| from subprocess import call |
| |
| from tests.common.environ import HIVE_MAJOR_VERSION |
| from tests.common.impala_test_suite import ImpalaTestSuite |
| from tests.common.skip import (SkipIfS3, SkipIfABFS, SkipIfADLS, SkipIfHive2, SkipIfHive3, |
| SkipIfIsilon, SkipIfLocal, SkipIfCatalogV2) |
| from tests.common.test_dimensions import ( |
| create_single_exec_option_dimension, |
| create_uncompressed_text_dimension) |
| from tests.util.hive_utils import HiveDbWrapper, HiveTableWrapper |
| |
| |
| @SkipIfS3.hive |
| @SkipIfABFS.hive |
| @SkipIfADLS.hive |
| @SkipIfIsilon.hive |
| @SkipIfLocal.hive |
| class TestHmsIntegrationSanity(ImpalaTestSuite): |
| @classmethod |
| def get_workload(self): |
| return 'functional-query' |
| |
| @classmethod |
| def add_test_dimensions(cls): |
| super(TestHmsIntegrationSanity, cls).add_test_dimensions() |
| # There is no reason to run these tests using all dimensions. |
| cls.ImpalaTestMatrix.add_dimension(create_single_exec_option_dimension()) |
| cls.ImpalaTestMatrix.add_dimension( |
| create_uncompressed_text_dimension(cls.get_workload())) |
| |
| @pytest.mark.execute_serially |
| def test_sanity(self, vector, cluster_properties): |
| """Verifies that creating a catalog entity (database, table) in Impala using |
| 'IF NOT EXISTS' while the entity exists in HMS, does not throw an error.""" |
| # Create a database in Hive |
| self.run_stmt_in_hive("drop database if exists hms_sanity_db cascade") |
| self.run_stmt_in_hive("create database hms_sanity_db") |
| # Make sure Impala's metadata is in sync. |
| # Invalidate metadata to pick up hive-created db. |
| self.client.execute("invalidate metadata") |
| # Creating a database with the same name using 'IF NOT EXISTS' in Impala should |
| # not fail |
| self.client.execute("create database if not exists hms_sanity_db") |
| # The database should appear in the catalog (IMPALA-2441) |
| assert 'hms_sanity_db' in self.all_db_names() |
| # Ensure a table can be created in this database from Impala and that it is |
| # accessable in both Impala and Hive |
| self.client.execute("create table hms_sanity_db.test_tbl_in_impala(a int)") |
| self.run_stmt_in_hive("select * from hms_sanity_db.test_tbl_in_impala") |
| self.client.execute("select * from hms_sanity_db.test_tbl_in_impala") |
| # Create a table in Hive |
| self.run_stmt_in_hive("create table hms_sanity_db.test_tbl (a int)") |
| # Creating a table with the same name using 'IF NOT EXISTS' in Impala should |
| # not fail |
| self.client.execute("create table if not exists hms_sanity_db.test_tbl (a int)") |
| # The table should not appear in the catalog *immediately* unless invalidate |
| # metadata is executed. |
| assert 'test_tbl' not in self.client.execute("show tables in hms_sanity_db").data |
| self.client.execute("invalidate metadata hms_sanity_db.test_tbl") |
| assert 'test_tbl' in self.client.execute("show tables in hms_sanity_db").data |
| |
| @SkipIfS3.hive |
| @SkipIfABFS.hive |
| @SkipIfADLS.hive |
| @SkipIfIsilon.hive |
| @SkipIfLocal.hive |
| class TestHmsIntegration(ImpalaTestSuite): |
| |
| @classmethod |
| def get_workload(self): |
| return 'functional-query' |
| |
| @classmethod |
| def add_test_dimensions(cls): |
| super(TestHmsIntegration, cls).add_test_dimensions() |
| |
| if cls.exploration_strategy() != 'exhaustive': |
| pytest.skip("Should only run in exhaustive due to long execution time.") |
| |
| # There is no reason to run these tests using all dimensions. |
| cls.ImpalaTestMatrix.add_dimension(create_single_exec_option_dimension()) |
| cls.ImpalaTestMatrix.add_dimension( |
| create_uncompressed_text_dimension(cls.get_workload())) |
| |
| class ImpalaDbWrapper(object): |
| """ |
| A wrapper class for using `with` guards with databases created through |
| Impala ensuring deletion even if an exception occurs. |
| """ |
| |
| def __init__(self, impala, db_name): |
| self.impala = impala |
| self.db_name = db_name |
| |
| def __enter__(self): |
| self.impala.client.execute( |
| 'create database if not exists ' + self.db_name) |
| return self.db_name |
| |
| def __exit__(self, typ, value, traceback): |
| self.impala.cleanup_db(self.db_name) |
| |
| class ImpalaTableWrapper(object): |
| """ |
| A wrapper class for using `with` guards with tables created through Impala |
| ensuring deletion even if an exception occurs. |
| """ |
| |
| def __init__(self, impala, table_name, table_spec): |
| self.impala = impala |
| self.table_name = table_name |
| self.table_spec = table_spec |
| |
| def __enter__(self): |
| self.impala.client.execute( |
| 'create table if not exists %s %s' % |
| (self.table_name, self.table_spec)) |
| return self.table_name |
| |
| def __exit__(self, typ, value, traceback): |
| self.impala.client.execute('drop table if exists %s' % self.table_name) |
| |
| def impala_table_stats(self, table): |
| """Returns a dictionary of stats for a table according to Impala.""" |
| output = self.client.execute('show table stats %s' % table).get_data() |
| output_lines = output.split('\n') |
| result = {} |
| for line in output_lines: |
| parts = line.split('\t') |
| stats = {} |
| stats['location'] = parts[-1] |
| stats['incremental stats'] = parts[-2] |
| stats['format'] = parts[-3] |
| stats['cache replication'] = parts[-4] |
| stats['bytes cached'] = parts[-5] |
| stats['size'] = parts[-6] |
| stats['#files'] = parts[-7] |
| stats['#rows'] = parts[-8] |
| result[tuple(parts[:-8])] = stats |
| return result |
| |
| def impala_all_column_stats(self, table): |
| """Returns a dictionary of stats for columns according to Impala.""" |
| output = self.client.execute('show column stats %s' % table).get_data() |
| output_lines = output.split('\n') |
| result = {} |
| for line in output_lines: |
| stats = line.split('\t') |
| attributes = {} |
| attributes['type'] = stats[1] |
| attributes['ndv'] = stats[2] |
| attributes['#nulls'] = stats[3] |
| attributes['max size'] = stats[4] |
| attributes['avg size'] = stats[5] |
| result[stats[0]] = attributes |
| return result |
| |
| def parse_hive2_describe_formatted_output(self, output): |
| """Parses the output of a 'describe formatted' statement for Hive 2. Returns a |
| dictionary that holds the parsed attributes.""" |
| result = {} |
| output_lines = output.split('\n') |
| stat_names = map(string.strip, output_lines[0].split(',')) |
| stat_values = output_lines[3].split(',') |
| assert len(stat_names) == len(stat_values) |
| for i in range(0, len(stat_names)): |
| result[stat_names[i]] = stat_values[i] |
| return result |
| |
| def parse_hive3_describe_formatted_output(self, output): |
| """Parses the output of a 'describe formatted' statement for Hive 3. Returns a |
| dictionary that holds the parsed attributes.""" |
| result = {} |
| for line in output.split('\n'): |
| line_elements = map(string.strip, line.split(',')) |
| if len(line_elements) >= 2: |
| result[line_elements[0]] = line_elements[1] |
| return result |
| |
| def hive_column_stats(self, table, column, remove_stats_accurate=False): |
| """Returns a dictionary of stats for a column according to Hive.""" |
| output = self.run_stmt_in_hive('describe formatted %s %s' % (table, column)) |
| result = {} |
| if HIVE_MAJOR_VERSION == 2: |
| result = self.parse_hive2_describe_formatted_output(output) |
| else: |
| result = self.parse_hive3_describe_formatted_output(output) |
| if remove_stats_accurate: |
| result.pop('COLUMN_STATS_ACCURATE', None) |
| return result |
| |
| def impala_columns(self, table_name): |
| """ |
| Returns a dict with column names as the keys and dicts of type and comments |
| as the values. |
| """ |
| columns = self.client.execute('describe %s' % |
| table_name).get_data().split('\n') |
| result = {} |
| for column in columns: |
| attributes = column.split('\t') |
| result[attributes[0]] = {'type': attributes[1], 'comment': attributes[2]} |
| return result |
| |
| def hive_columns(self, table_name): |
| """ |
| Returns a dict with column names as the keys and dicts of types and |
| comments as the values. |
| """ |
| |
| columns = self.run_stmt_in_hive( |
| 'describe %s' % table_name).split('\n')[1:-1] |
| result = {} |
| for column in columns: |
| attributes = column.split(',') |
| result[attributes[0]] = {'type': attributes[1], 'comment': attributes[2]} |
| return result |
| |
| def unique_string(self): |
| return ''.join([random.choice(string.ascii_lowercase) |
| for i in range(0, 16)]) |
| |
| def assert_sql_error(self, engine, command, *strs_in_error): |
| """ |
| Passes 'command' to 'engine' callable (e.g. execute method of a BeeswaxConnection |
| object) and makes sure that it raises an exception. |
| It also verifies that the string representation of the exception contains all the |
| strings listed in 'strs_in_error'. |
| |
| If the call doesn't raise an exception or the exception doesn't contain one of the |
| strings in 'strs_in_error', it throws AssertError exception. |
| """ |
| |
| try: |
| engine(command) |
| except Exception as e: |
| for str_in_error in strs_in_error: |
| assert str_in_error in str(e) |
| else: |
| assert False, '%s should have triggered an error containing %s' % ( |
| command, strs_in_error) |
| |
| @pytest.mark.execute_serially |
| def test_hive_db_hive_table_add_partition(self, vector): |
| self.add_hive_partition_helper(vector, HiveDbWrapper, HiveTableWrapper) |
| |
| @pytest.mark.execute_serially |
| def test_hive_db_impala_table_add_partition(self, vector): |
| self.add_hive_partition_helper(vector, HiveDbWrapper, self.ImpalaTableWrapper) |
| |
| @pytest.mark.execute_serially |
| def test_impala_db_impala_table_add_partition(self, vector): |
| self.add_hive_partition_helper(vector, self.ImpalaDbWrapper, |
| self.ImpalaTableWrapper) |
| |
| @pytest.mark.execute_serially |
| def test_impala_db_hive_table_add_partition(self, vector): |
| self.add_hive_partition_helper(vector, self.ImpalaDbWrapper, |
| HiveTableWrapper) |
| |
| @pytest.mark.xfail(run=False, reason="This is a bug: IMPALA-2426") |
| @pytest.mark.execute_serially |
| def test_incremental_stats_new_partition(self, vector): |
| with self.ImpalaDbWrapper(self, self.unique_string()) as db_name: |
| with self.ImpalaTableWrapper(self, db_name + '.' + self.unique_string(), |
| '(x int) partitioned by (y int)') as table_name: |
| self.client.execute('insert into table %s partition (y=42) values (2)' |
| % table_name) |
| self.run_stmt_in_hive('alter table %s add partition (y = 333)' |
| % table_name) |
| self.client.execute('compute incremental stats %s' % table_name) |
| table_stats = self.impala_table_stats(table_name) |
| assert 'true' == table_stats[('333',)]['incremental stats'] |
| assert '0' == table_stats[('333',)]['#rows'] |
| assert '0' == table_stats[('333',)]['#files'] |
| |
| def add_hive_partition_helper(self, vector, DbWrapper, TableWrapper): |
| """ |
| Partitions added in Hive can be viewed in Impala after computing stats. |
| """ |
| with DbWrapper(self, self.unique_string()) as db_name: |
| self.client.execute('invalidate metadata') |
| with TableWrapper(self, db_name + '.' + self.unique_string(), |
| '(x int) partitioned by (y int, z int)') as table_name: |
| # Invalidate metadata so Impala can see the table |
| self.client.execute('invalidate metadata') |
| self.run_stmt_in_hive( |
| 'alter table %s add partition (y = 333, z = 5309)' % |
| table_name) |
| self.client.execute('compute incremental stats %s' % table_name) |
| # Impala can see the partition's name |
| assert [('333', '5309')] == self.get_impala_partition_info(table_name, 'y', 'z') |
| # Impala's compute stats didn't alter Hive's knowledge of the partition |
| assert ['y=333/z=5309'] == self.hive_partition_names(table_name) |
| self.add_hive_partition_table_stats_helper(vector, DbWrapper, TableWrapper) |
| |
| def add_hive_partition_table_stats_helper( |
| self, vector, DbWrapper, TableWrapper): |
| """ |
| Partitions added in Hive don't make Impala's table stats incorrect. |
| """ |
| # TODO: check the same thing with column stats |
| with DbWrapper(self, self.unique_string()) as db_name: |
| self.client.execute('invalidate metadata') |
| with TableWrapper(self, db_name + '.' + self.unique_string(), |
| '(x int) partitioned by (y int, z int)') as table_name: |
| # Invalidate metadata so Impala can see the table |
| self.client.execute('invalidate metadata') |
| self.client.execute( |
| 'insert into table %s partition (y=42, z=867) values (2)' |
| % table_name) |
| self.client.execute('compute incremental stats %s' % table_name) |
| impala_table_stats = self.impala_table_stats(table_name) |
| self.run_stmt_in_hive( |
| 'alter table %s add partition (y = 333, z = 5309)' % |
| table_name) |
| self.client.execute('compute incremental stats %s' % table_name) |
| assert impala_table_stats[ |
| ('42', '867')] == self.impala_table_stats(table_name)[ |
| ('42', '867')] |
| |
| @pytest.mark.execute_serially |
| def test_add_impala_partition(self, vector): |
| """ |
| Partitions added in Impala can be viewed in Hive immediately |
| """ |
| with self.ImpalaDbWrapper(self, self.unique_string()) as db_name: |
| with self.ImpalaTableWrapper(self, db_name + '.' + self.unique_string(), |
| '(x int) partitioned by (y int, z int)') as table_name: |
| self.client.execute( |
| 'insert into table %s partition (y=42, z=867) values (2)' |
| % table_name) |
| assert [('42', '867')] == self.get_impala_partition_info(table_name, 'y', 'z') |
| assert ['y=42/z=867'] == self.hive_partition_names(table_name) |
| |
| @pytest.mark.execute_serially |
| def test_drop_column_maintains_stats(self, vector): |
| """ |
| Dropping a column in Impala doesn't alter the stats of other columns in Hive |
| or Impala. |
| """ |
| with self.ImpalaDbWrapper(self, self.unique_string()) as db_name: |
| with self.ImpalaTableWrapper(self, db_name + '.' + self.unique_string(), |
| '(x int, y int, z int)') as table_name: |
| self.run_stmt_in_hive('select * from %s' % table_name) |
| self.run_stmt_in_hive( |
| 'use %s; analyze table %s compute statistics for columns' % |
| (db_name, table_name.split('.')[1])) |
| self.client.execute('compute stats %s' % table_name) |
| hive_x_stats = self.hive_column_stats(table_name, 'x') |
| hive_y_stats = self.hive_column_stats(table_name, 'y') |
| impala_stats = self.impala_all_column_stats(table_name) |
| self.client.execute('alter table %s drop column z' % table_name) |
| assert hive_x_stats == self.hive_column_stats(table_name, 'x', |
| remove_stats_accurate=True) |
| assert hive_y_stats == self.hive_column_stats(table_name, 'y', |
| remove_stats_accurate=True) |
| assert impala_stats['x'] == self.impala_all_column_stats(table_name)[ |
| 'x'] |
| assert impala_stats['y'] == self.impala_all_column_stats(table_name)[ |
| 'y'] |
| self.run_stmt_in_hive( |
| 'alter table %s replace columns (x int)' % |
| table_name) |
| assert hive_x_stats == self.hive_column_stats(table_name, 'x', |
| remove_stats_accurate=True) |
| assert impala_stats['x'] == self.impala_all_column_stats(table_name)[ |
| 'x'] |
| |
| @pytest.mark.execute_serially |
| def test_select_without_compute_stats(self, vector): |
| """ |
| Data added in Hive shows up in Impala 'select *', and if the table is not |
| partitioned, 'compute incremental stats' is not required. |
| """ |
| |
| with self.ImpalaDbWrapper(self, self.unique_string()) as db_name: |
| with self.ImpalaTableWrapper(self, db_name + '.' + self.unique_string(), |
| '(x int)') as table_name: |
| # In the unpartitioned case, 'compute incremental stats' is not |
| # required. |
| self.run_stmt_in_hive( |
| 'insert into table %s values (66)' |
| % table_name) |
| assert '66' == self.client.execute( |
| 'select * from %s' % table_name).get_data() |
| |
| with self.ImpalaTableWrapper(self, db_name + '.' + self.unique_string(), |
| '(x int) partitioned by (y int)') as table_name: |
| assert [] == self.get_impala_partition_info(table_name, 'y') |
| self.run_stmt_in_hive( |
| 'insert into table %s partition (y=33) values (44)' |
| % table_name) |
| self.client.execute('compute incremental stats %s' % table_name) |
| assert '44\t33' == self.client.execute( |
| 'select * from %s' % table_name).get_data() |
| |
| @pytest.mark.xfail(run=False, reason="This is a bug: IMPALA-2458") |
| @pytest.mark.execute_serially |
| def test_overwrite_added_column(self, vector): |
| """ |
| Impala can't overwrite Hive's column types, and vice versa. |
| """ |
| with self.ImpalaDbWrapper(self, self.unique_string()) as db_name: |
| with self.ImpalaTableWrapper(self, db_name + '.' + self.unique_string(), |
| '(x int, y int)') as table_name: |
| inttype = {'comment': '', 'type': 'int'} |
| hive_expected = {'x': inttype, 'y': inttype} |
| impala_expected = {'x': inttype, 'y': inttype} |
| # Hive and Impala both know all columns: |
| assert hive_expected == self.hive_columns(table_name) |
| assert impala_expected == self.impala_columns(table_name) |
| # Add column in Hive but don't tell Impala |
| self.run_stmt_in_hive( |
| 'alter table %s add columns (z int)' % table_name) |
| hive_expected['z'] = inttype |
| assert hive_expected == self.hive_columns(table_name) |
| # Overwriting an Hive-created column in Impala does not work |
| self.assert_sql_error( |
| self.client.execute, |
| 'alter table %s add columns (z string)' % |
| table_name, |
| 'Column already exists: z') |
| # Overwriting an Impala-created column in Hive does not work |
| self.client.execute( |
| 'alter table %s add columns (v string)' % table_name) |
| self.assert_sql_error( |
| self.run_stmt_in_hive, |
| 'alter table %s add columns (v string)' % |
| table_name, |
| 'Duplicate column name: v') |
| |
| @SkipIfHive3.col_stat_separated_by_engine |
| @pytest.mark.execute_serially |
| def test_compute_stats_get_to_hive(self, vector): |
| """Stats computed in Impala are also visible in Hive.""" |
| with self.ImpalaDbWrapper(self, self.unique_string()) as db_name: |
| with self.ImpalaTableWrapper(self, db_name + '.' + self.unique_string(), |
| '(x int)') as table_name: |
| self.run_stmt_in_hive( |
| 'insert into table %s values (33)' % table_name) |
| hive_stats = self.hive_column_stats(table_name, 'x') |
| impala_stats = self.client.execute('show column stats %s' % table_name) |
| self.client.execute('compute stats %s' % table_name) |
| assert impala_stats != self.client.execute( |
| 'show column stats %s' % table_name) |
| assert hive_stats != self.hive_column_stats(table_name, 'x') |
| |
| @SkipIfHive3.col_stat_separated_by_engine |
| @pytest.mark.execute_serially |
| def test_compute_stats_get_to_impala(self, vector): |
| """Column stats computed in Hive are also visible in Impala.""" |
| with HiveDbWrapper(self, self.unique_string()) as db_name: |
| with HiveTableWrapper(self, db_name + '.' + self.unique_string(), |
| '(x int)') as table_name: |
| hive_stats = self.hive_column_stats(table_name, 'x') |
| self.client.execute('invalidate metadata') |
| self.client.execute('refresh %s' % table_name) |
| impala_stats = self.impala_all_column_stats(table_name) |
| self.run_stmt_in_hive( |
| 'insert into table %s values (33)' % table_name) |
| self.run_stmt_in_hive( |
| 'use %s; analyze table %s compute statistics for columns' % |
| (db_name, table_name.split('.')[1])) |
| new_hive_stats = self.hive_column_stats(table_name, 'x') |
| assert hive_stats != new_hive_stats |
| assert '33' == new_hive_stats['min'] |
| assert '33' == new_hive_stats['max'] |
| assert '0' == new_hive_stats['num_nulls'] |
| self.client.execute('refresh %s' % table_name) |
| new_impala_stats = self.impala_all_column_stats(table_name) |
| assert impala_stats != new_impala_stats |
| assert '0' == new_impala_stats['x']['#nulls'] |
| |
| @SkipIfHive2.col_stat_not_separated_by_engine |
| def test_engine_separates_col_stats(self, vector): |
| """ |
| The 'engine' column in TAB_COL_STATS and PART_COL_STATS HMS tables is used to |
| differentiate among column stats computed by different engines. |
| |
| IMPALA-8842: Test that Impala sets 'engine' column correctly when writing/reading |
| column statistics for a non-partitioned table. Both Hive and Impala use TAB_COL_STATS |
| to store the non-partitioned table column statistics. |
| |
| The test is executed for transactional and non-transactional tables. |
| """ |
| trans_tbl_prop = \ |
| "TBLPROPERTIES('transactional'='true', 'transactional_properties'='insert_only')" |
| for tbl_prop in [trans_tbl_prop, '']: |
| with self.ImpalaDbWrapper(self, self.unique_string()) as db_name: |
| with self.ImpalaTableWrapper(self, '%s.%s' % (db_name, self.unique_string()), |
| '(x int) %s' % tbl_prop) as table_name: |
| self.run_stmt_in_hive('insert into %s values (0), (1), (2), (3)' % table_name) |
| self.run_stmt_in_hive( |
| 'use %s; analyze table %s compute statistics for columns' % |
| (db_name, table_name.split('.')[1])) |
| hive_x_stats = self.hive_column_stats(table_name, 'x') |
| assert '4' == hive_x_stats['distinct_count'] |
| assert '0' == hive_x_stats['num_nulls'] |
| |
| # Impala doesn't read column stats written by Hive. |
| self.client.execute('invalidate metadata %s' % table_name) |
| impala_stats = self.impala_all_column_stats(table_name) |
| assert '-1' == impala_stats['x']['#nulls'] |
| assert '-1' == impala_stats['x']['ndv'] |
| # Impala writes and reads its own column stats |
| self.client.execute('compute stats %s' % table_name) |
| impala_stats = self.impala_all_column_stats(table_name) |
| assert '0' == impala_stats['x']['#nulls'] |
| assert '4' == impala_stats['x']['ndv'] |
| # Insert additional rows and recalculate stats in Impala. |
| self.client.execute('insert into %s values (10), (11), (12), (13)' % table_name) |
| self.client.execute('compute stats %s' % table_name) |
| impala_stats = self.impala_all_column_stats(table_name) |
| assert '0' == impala_stats['x']['#nulls'] |
| assert '8' == impala_stats['x']['ndv'] |
| |
| # Hive doesn't read column stats written by Impala |
| hive_x_stats = self.hive_column_stats(table_name, 'x') |
| assert '4' == hive_x_stats['distinct_count'] |
| assert '0' == hive_x_stats['num_nulls'] |
| |
| @SkipIfHive2.col_stat_not_separated_by_engine |
| def test_engine_separates_partitioned_col_stats(self, vector): |
| """ |
| The 'engine' column in TAB_COL_STATS and PART_COL_STATS HMS tables is used to |
| differentiate among column stats computed by different engines. |
| |
| IMPALA-8842: Test that Impala sets 'engine' column correctly when writing/reading |
| column statistics for a partitioned table. Note, that Hive uses PART_COL_STATS to |
| store parttioned table column statistics whereas Impala stores them in TAB_COL_STATS, |
| therefore column stats would have been separated even without IMPALA-8842. |
| |
| The test is executed for transactional and non-transactional tables. |
| """ |
| trans_tbl_prop = \ |
| "TBLPROPERTIES('transactional'='true', 'transactional_properties'='insert_only')" |
| for tbl_prop in [trans_tbl_prop, '']: |
| with self.ImpalaDbWrapper(self, self.unique_string()) as db_name: |
| with self.ImpalaTableWrapper(self, '%s.%s' % (db_name, self.unique_string()), |
| '(x int) partitioned by (y int) %s' % tbl_prop) as table_name: |
| self.run_stmt_in_hive( |
| 'insert into %s partition (y=0) values (0), (1), (2), (3)' % table_name) |
| self.run_stmt_in_hive( |
| 'use %s; analyze table %s compute statistics for columns' % |
| (db_name, table_name.split('.')[1])) |
| hive_x_stats = self.hive_column_stats(table_name, 'x') |
| assert '4' == hive_x_stats['distinct_count'] |
| assert '0' == hive_x_stats['num_nulls'] |
| |
| # Impala doesn't read column stats written by Hive. |
| self.client.execute('invalidate metadata %s' % table_name) |
| impala_stats = self.impala_all_column_stats(table_name) |
| assert '-1' == impala_stats['x']['#nulls'] |
| assert '-1' == impala_stats['x']['ndv'] |
| # Impala writes and reads its own column stats |
| self.client.execute('compute stats %s' % table_name) |
| impala_stats = self.impala_all_column_stats(table_name) |
| assert '0' == impala_stats['x']['#nulls'] |
| assert '4' == impala_stats['x']['ndv'] |
| # Insert additional rows and recalculate stats in Impala. |
| self.client.execute( |
| 'insert into %s partition (y=0) values (10), (11), (12), (13)' % table_name) |
| self.client.execute('compute stats %s' % table_name) |
| impala_stats = self.impala_all_column_stats(table_name) |
| assert '0' == impala_stats['x']['#nulls'] |
| assert '8' == impala_stats['x']['ndv'] |
| |
| # Hive doesn't read column stats written by Impala |
| hive_x_stats = self.hive_column_stats(table_name, 'x') |
| assert '4' == hive_x_stats['distinct_count'] |
| assert '0' == hive_x_stats['num_nulls'] |
| |
| @pytest.mark.execute_serially |
| def test_drop_partition(self, vector): |
| """ |
| Impala can see that a partitions was dropped by Hive by invalidating |
| metadata. |
| """ |
| |
| with self.ImpalaDbWrapper(self, self.unique_string()) as db_name: |
| with self.ImpalaTableWrapper(self, db_name + '.' + self.unique_string(), |
| '(x int) partitioned by (y int)') as table_name: |
| self.run_stmt_in_hive( |
| 'insert into table %s partition(y=33) values (44)' % table_name) |
| self.client.execute('compute stats %s' % table_name) |
| self.run_stmt_in_hive( |
| 'alter table %s drop partition (y=33)' % table_name) |
| self.client.execute('invalidate metadata %s' % table_name) |
| assert '' == self.client.execute( |
| 'select * from %s' % table_name).get_data() |
| |
| @pytest.mark.execute_serially |
| def test_drop_column_with_data(self, vector): |
| """Columns dropped by Hive are ignored in Impala 'select *'.""" |
| with self.ImpalaDbWrapper(self, self.unique_string()) as db_name: |
| with self.ImpalaTableWrapper(self, db_name + '.' + self.unique_string(), |
| '(x int, y int)') as table_name: |
| self.run_stmt_in_hive( |
| 'insert into table %s values (33,44)' % table_name) |
| self.run_stmt_in_hive( |
| 'alter table %s replace columns (x int)' % table_name) |
| assert '33' == self.client.execute( |
| 'select * from %s' % table_name).get_data() |
| |
| @pytest.mark.execute_serially |
| def test_add_column(self, vector): |
| """Columns added in one engine are visible in the other via DESCRIBE.""" |
| with self.ImpalaDbWrapper(self, self.unique_string()) as db_name: |
| with self.ImpalaTableWrapper(self, db_name + '.' + self.unique_string(), |
| '(x int)') as table_name: |
| int_column = {'type': 'int', 'comment': ''} |
| expected = {'x': int_column} |
| assert expected == self.hive_columns(table_name) |
| assert expected == self.impala_columns(table_name) |
| self.client.execute('alter table %s add columns (y int)' % table_name) |
| expected['y'] = int_column |
| assert expected == self.hive_columns(table_name) |
| assert expected == self.impala_columns(table_name) |
| self.run_stmt_in_hive( |
| 'alter table %s add columns (z int)' % |
| table_name) |
| self.client.execute('invalidate metadata %s' % table_name) |
| expected['z'] = int_column |
| assert expected == self.hive_columns(table_name) |
| assert expected == self.impala_columns(table_name) |
| |
| @pytest.mark.execute_serially |
| def test_drop_database(self, vector): |
| """ |
| If a DB is created, then dropped, in Hive, Impala can create one with the |
| same name without invalidating metadata. |
| """ |
| |
| test_db = self.unique_string() |
| with HiveDbWrapper(self, test_db) as db_name: |
| pass |
| self.assert_sql_error( |
| self.client.execute, |
| 'create table %s.%s (x int)' % |
| (test_db, |
| self.unique_string()), |
| 'Database does not exist: %s' % |
| test_db) |
| with self.ImpalaDbWrapper(self, test_db) as db_name: |
| pass |
| |
| @pytest.mark.execute_serially |
| def test_table_format_change(self, vector): |
| """ |
| Hive storage format changes propagate to Impala. |
| """ |
| # TODO: check results of insert, then select * before and after |
| # storage format change. |
| with HiveDbWrapper(self, self.unique_string()) as db_name: |
| with HiveTableWrapper(self, db_name + '.' + self.unique_string(), |
| '(x int, y int) stored as parquet') as table_name: |
| self.client.execute('invalidate metadata') |
| self.client.execute('invalidate metadata %s' % table_name) |
| print self.impala_table_stats(table_name) |
| assert 'PARQUET' == self.impala_table_stats(table_name)[()]['format'] |
| self.run_stmt_in_hive( |
| 'alter table %s set fileformat avro' % table_name) |
| self.client.execute('invalidate metadata %s' % table_name) |
| assert 'AVRO' == self.impala_table_stats(table_name)[()]['format'] |
| |
| @pytest.mark.execute_serially |
| def test_change_column_type(self, vector): |
| """Hive column type changes propagate to Impala.""" |
| |
| with HiveDbWrapper(self, self.unique_string()) as db_name: |
| with HiveTableWrapper(self, db_name + '.' + self.unique_string(), |
| '(x int, y int)') as table_name: |
| self.run_stmt_in_hive( |
| 'insert into table %s values (33,44)' % table_name) |
| self.run_stmt_in_hive('alter table %s change y y string' % table_name) |
| assert '33,44' == self.run_stmt_in_hive( |
| 'select * from %s' % table_name).split('\n')[1] |
| self.client.execute('invalidate metadata %s' % table_name) |
| assert '33\t44' == self.client.execute( |
| 'select * from %s' % table_name).get_data() |
| assert 'string' == self.impala_columns(table_name)['y']['type'] |
| |
| @pytest.mark.execute_serially |
| def test_change_parquet_column_type(self, vector): |
| """ |
| Changing column types in Parquet doesn't always work in Hive and it causes |
| 'select *' to fail in Impala as well, after invalidating metadata. This is a |
| known issue with changing column types in Hive/parquet. |
| """ |
| |
| with HiveDbWrapper(self, self.unique_string()) as db_name: |
| with HiveTableWrapper(self, db_name + '.' + self.unique_string(), |
| '(x int, y int) stored as parquet') as table_name: |
| # The following INSERT statement creates a Parquet file with INT columns. |
| self.run_stmt_in_hive('insert into table %s values (33,44)' % table_name) |
| assert '33,44' == self.run_stmt_in_hive( |
| 'select * from %s' % table_name).split('\n')[1] |
| self.client.execute('invalidate metadata') |
| assert '33\t44' == self.client.execute( |
| 'select * from %s' % table_name).get_data() |
| # Modify table metadata. After this statement, the table metadata in HMS |
| # and the Parquet file metadata won't agree on the type of column 'y'. |
| self.run_stmt_in_hive('alter table %s change y y string' % table_name) |
| if HIVE_MAJOR_VERSION == 2: |
| # Hive 2 doesn't allow implicit conversion from INT to STRING. |
| self.assert_sql_error( |
| self.run_stmt_in_hive, 'select * from %s' % table_name, |
| 'Cannot inspect org.apache.hadoop.io.IntWritable') |
| else: |
| # Hive 3 implicitly converts INTs to STRINGs. |
| assert '33,44' == self.run_stmt_in_hive( |
| 'select * from %s' % table_name).split('\n')[1] |
| self.client.execute('invalidate metadata %s' % table_name) |
| # Impala doesn't convert INTs to STRINGs implicitly. |
| self.assert_sql_error( |
| self.client.execute, 'select * from %s' % table_name, |
| "Column type: STRING, Parquet schema:") |
| # Insert STRING value, it will create a Parquet file where column 'y' |
| # has type STRING. |
| self.run_stmt_in_hive('insert into table %s values (33,\'100\')' % table_name) |
| # Modify HMS table metadata again, change the type of column 'y' back to INT. |
| self.run_stmt_in_hive('alter table %s change y y int' % table_name) |
| # Neither Hive 2 and 3, nor Impala converts STRINGs to INTs implicitly. |
| self.assert_sql_error( |
| self.run_stmt_in_hive, 'select * from %s' % table_name, |
| 'org.apache.hadoop.io.Text cannot be ' |
| 'cast to org.apache.hadoop.io.IntWritable') |
| self.client.execute('invalidate metadata %s' % table_name) |
| self.assert_sql_error( |
| self.client.execute, 'select * from %s' % table_name, |
| "Column type: INT, Parquet schema:") |
| |
| @SkipIfHive2.acid |
| def test_acid_inserts(self, vector, unique_database): |
| """ |
| Insert data to insert-only ACID table from Impala and checks that Hive is able to |
| see the data. |
| """ |
| table_name = "%s.acid_insert" % unique_database |
| self.client.execute( |
| "create table %s (i int) " |
| "TBLPROPERTIES('transactional'='true', " |
| "'transactional_properties'='insert_only')" % table_name) |
| self.client.execute("insert into table %s values (1)" % table_name) |
| assert '1' == self.run_stmt_in_hive("select * from %s" % table_name).split('\n')[1] |
| self.client.execute("insert into table %s values (2)" % table_name) |
| assert '2' == self.run_stmt_in_hive( |
| "select * from %s order by i" % table_name).split('\n')[2] |
| self.client.execute("insert overwrite table %s values (10)" % table_name) |
| assert '10' == self.run_stmt_in_hive("select * from %s" % table_name).split('\n')[1] |
| self.client.execute("insert into table %s values (11)" % table_name) |
| assert '11' == self.run_stmt_in_hive( |
| "select * from %s order by i" % table_name).split('\n')[2] |
| assert '2' == self.run_stmt_in_hive("select count(*) from %s" |
| % table_name).split('\n')[1] |
| |
| # CTAS ACID table with Impala and select from Hive |
| ctas_table_name = "%s.acid_insert_ctas" % unique_database |
| self.client.execute( |
| "create table %s " |
| "TBLPROPERTIES('transactional'='true', " |
| "'transactional_properties'='insert_only') " |
| "as select * from %s" % (ctas_table_name, table_name)) |
| assert '11' == self.run_stmt_in_hive( |
| "select * from %s order by i" % ctas_table_name).split('\n')[2] |
| assert '2' == self.run_stmt_in_hive("select count(*) from %s" |
| % ctas_table_name).split('\n')[1] |
| |
| # Insert into partitioned ACID table |
| part_table_name = "%s.part_acid_insert" % unique_database |
| self.client.execute( |
| "create table %s (i int) partitioned by (p int)" |
| "TBLPROPERTIES('transactional'='true', " |
| "'transactional_properties'='insert_only')" % part_table_name) |
| self.client.execute("insert into %s partition (p=1) values (10)" % part_table_name) |
| self.client.execute("insert into %s partition (p=2) values (20)" % part_table_name) |
| hive_result = self.run_stmt_in_hive( |
| "select p, i from %s order by p, i" % part_table_name).split('\n') |
| assert '1,10' == hive_result[1] |
| assert '2,20' == hive_result[2] |
| self.client.execute( |
| "insert into %s partition (p) values (30,3),(40,4)" % part_table_name) |
| hive_result = self.run_stmt_in_hive( |
| "select p, i from %s order by p, i" % part_table_name).split('\n') |
| assert '3,30' == hive_result[3] |
| assert '4,40' == hive_result[4] |
| self.client.execute( |
| "insert overwrite %s partition (p) values (11,1),(41,4)" % part_table_name) |
| hive_result = self.run_stmt_in_hive( |
| "select p, i from %s order by p, i" % part_table_name).split('\n') |
| assert '1,11' == hive_result[1] |
| assert '2,20' == hive_result[2] |
| assert '3,30' == hive_result[3] |
| assert '4,41' == hive_result[4] |
| |
| @SkipIfHive2.acid |
| def test_drop_acid_table(self, vector, unique_database): |
| """ |
| Tests that a transactional table dropped by Impala is also dropped if we check from |
| Hive. |
| """ |
| table_name = "%s.acid_insert" % unique_database |
| self.client.execute( |
| "create table %s (i int) " |
| "TBLPROPERTIES('transactional'='true', " |
| "'transactional_properties'='insert_only')" % table_name) |
| show_tables_result = self.run_stmt_in_hive("show tables in %s" % unique_database) |
| assert "acid_insert" in show_tables_result |
| self.client.execute("drop table %s" % table_name) |
| show_tables_result_after_drop = self.run_stmt_in_hive( |
| "show tables in %s" % unique_database) |
| assert "acid_insert" not in show_tables_result_after_drop |
| |
| @SkipIfHive2.acid |
| def test_truncate_acid_table(self, vector, unique_database): |
| """ |
| Tests that a transactional table truncated by Impala shows no rows when |
| queried by Hive. |
| """ |
| table_name = "%s.acid_truncate" % unique_database |
| self.client.execute( |
| "create table %s (i int) " |
| "TBLPROPERTIES('transactional'='true', " |
| "'transactional_properties'='insert_only')" % table_name) |
| self.client.execute("insert into %s values (1), (2)" % table_name) |
| query_result = self.run_stmt_in_hive("select * from %s" % table_name) |
| assert "1" in query_result |
| assert "2" in query_result |
| self.client.execute("truncate table %s" % table_name) |
| query_result_after_truncate = self.run_stmt_in_hive("select count(*) from %s" % |
| table_name) |
| assert "0" == query_result_after_truncate.split('\n')[1] |
| |
| @pytest.mark.execute_serially |
| def test_change_table_name(self, vector): |
| """ |
| Changing the table name in Hive propagates to Impala after 'invalidate |
| metadata'. |
| """ |
| |
| with HiveDbWrapper(self, self.unique_string()) as db_name: |
| with HiveTableWrapper(self, db_name + '.' + self.unique_string(), |
| '(x int, y int)') as table_name: |
| self.client.execute('invalidate metadata') |
| int_column = {'type': 'int', 'comment': ''} |
| expected_columns = {'x': int_column, 'y': int_column} |
| assert expected_columns == self.impala_columns(table_name) |
| new_name = table_name + '2' |
| self.run_stmt_in_hive('alter table %s rename to %s' % |
| (table_name, new_name)) |
| self.client.execute('invalidate metadata') |
| assert expected_columns == self.impala_columns(new_name) |
| self.assert_sql_error(self.client.execute, |
| 'describe %s' % table_name, |
| 'Could not resolve path') |
| |
| @pytest.mark.execute_serially |
| def test_add_overlapping_partitions(self, vector): |
| """ |
| IMPALA-1670, IMPALA-4141: Test interoperability with Hive when adding overlapping |
| partitions to a table |
| """ |
| with self.ImpalaDbWrapper(self, self.unique_string()) as db_name: |
| # Create a table in Impala. |
| with self.ImpalaTableWrapper(self, db_name + '.' + self.unique_string(), |
| '(a int) partitioned by (x int)') as table_name: |
| # Trigger metadata load. No partitions exist yet in Impala. |
| assert [] == self.get_impala_partition_info(table_name, 'x') |
| |
| # Add partition in Hive. |
| self.run_stmt_in_hive("alter table %s add partition (x=2)" % table_name) |
| # Impala is not aware of the new partition. |
| assert [] == self.get_impala_partition_info(table_name, 'x') |
| |
| # Try to add partitions with caching in Impala, one of them (x=2) exists in HMS. |
| self.assert_sql_error(self.client.execute, |
| "alter table %s add partition (x=1) uncached " |
| "partition (x=2) cached in 'testPool' with replication=2 " |
| "partition (x=3) cached in 'testPool' with replication=3" % table_name, |
| "Partition already exists") |
| # No partitions were added in Impala. |
| assert [] == self.get_impala_partition_info(table_name, 'x') |
| |
| # It should succeed with IF NOT EXISTS. |
| self.client.execute("alter table %s add if not exists partition (x=1) uncached " |
| "partition (x=2) cached in 'testPool' with replication=2 " |
| "partition (x=3) cached in 'testPool' with replication=3" % table_name) |
| |
| # Hive sees all the partitions. |
| assert ['x=1', 'x=2', 'x=3'] == self.hive_partition_names(table_name) |
| |
| # Impala sees the partition that has already existed in HMS (x=2) and the newly |
| # added partitions (x=1) and (x=3). |
| # Caching has been applied only to newly added partitions (x=1) and (x=3), the |
| # preexisting partition (x=2) was not modified. |
| partitions = self.get_impala_partition_info(table_name, 'x', 'Bytes Cached', |
| 'Cache Replication') |
| assert [('1', 'NOT CACHED', 'NOT CACHED'), |
| ('2', 'NOT CACHED', 'NOT CACHED'), |
| ('3', '0B', '3')] == partitions |
| |
| # Try to add location to a partition that is already in catalog cache (x=1). |
| self.client.execute("alter table %s add if not exists "\ |
| "partition (x=1) location '/_X_1'" % table_name) |
| # (x=1) partition's location hasn't changed |
| (x1_value, x1_location) = self.get_impala_partition_info(table_name, 'x', |
| 'Location')[0] |
| assert '1' == x1_value |
| assert x1_location.endswith("/x=1"); |
| |
| @pytest.mark.execute_serially |
| def test_add_preexisting_partitions_with_data(self, vector): |
| """ |
| IMPALA-1670, IMPALA-4141: After addding partitions that already exist in HMS, Impala |
| can access the partition data. |
| """ |
| with self.ImpalaDbWrapper(self, self.unique_string()) as db_name: |
| # Create a table in Impala. |
| with self.ImpalaTableWrapper(self, db_name + '.' + self.unique_string(), |
| '(a int) partitioned by (x int)') as table_name: |
| # Trigger metadata load. No partitions exist yet in Impala. |
| assert [] == self.get_impala_partition_info(table_name, 'x') |
| |
| # Add partitions in Hive. |
| self.run_stmt_in_hive("alter table %s add partition (x=1) " |
| "partition (x=2) " |
| "partition (x=3)" % table_name) |
| # Insert rows in Hive |
| self.run_stmt_in_hive("insert into %s partition(x=1) values (1), (2), (3)" |
| % table_name) |
| self.run_stmt_in_hive("insert into %s partition(x=2) values (1), (2), (3), (4)" |
| % table_name) |
| self.run_stmt_in_hive("insert into %s partition(x=3) values (1)" |
| % table_name) |
| # No partitions exist yet in Impala. |
| assert [] == self.get_impala_partition_info(table_name, 'x') |
| |
| # Add the same partitions in Impala with IF NOT EXISTS. |
| self.client.execute("alter table %s add if not exists partition (x=1) "\ |
| "partition (x=2) " |
| "partition (x=3)" % table_name) |
| # Impala sees the partitions |
| assert [('1',), ('2',), ('3',)] == self.get_impala_partition_info(table_name, 'x') |
| # Data exists in Impala |
| assert ['1\t1', '1\t2', '1\t3', |
| '2\t1', '2\t2', '2\t3', '2\t4', |
| '3\t1'] ==\ |
| self.client.execute('select x, a from %s order by x, a' % |
| table_name).get_data().split('\n') |
| |
| @pytest.mark.execute_serially |
| def test_impala_partitions_accessible_in_hive(self, vector): |
| """ |
| IMPALA-1670, IMPALA-4141: Partitions added in Impala are accessible through Hive |
| """ |
| with self.ImpalaDbWrapper(self, self.unique_string()) as db_name: |
| # Create a table in Impala. |
| with self.ImpalaTableWrapper(self, db_name + '.' + self.unique_string(), |
| '(a int) partitioned by (x int)') as table_name: |
| # Add partitions in Impala. |
| self.client.execute("alter table %s add partition (x=1) " |
| "partition (x=2) " |
| "partition (x=3)" % table_name) |
| # Insert rows in Impala |
| self.client.execute("insert into %s partition(x=1) values (1), (2), (3)" |
| % table_name) |
| self.client.execute("insert into %s partition(x=2) values (1), (2), (3), (4)" |
| % table_name) |
| self.client.execute("insert into %s partition(x=3) values (1)" |
| % table_name) |
| |
| # Hive sees the partitions |
| assert ['x=1', 'x=2', 'x=3'] == self.hive_partition_names(table_name) |
| # Data exists in Hive |
| data = self.run_stmt_in_hive('select x, a from %s order by x, a' % table_name) |
| assert ['x,a', |
| '1,1', '1,2', '1,3', |
| '2,1', '2,2', '2,3', '2,4', |
| '3,1'] == data.strip().split('\n') |
