| # 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 __future__ import absolute_import, division, print_function |
| from builtins import range |
| import logging |
| from os import getenv |
| import pytest |
| |
| |
| from beeswaxd.BeeswaxService import QueryState |
| from hive_metastore.ttypes import FireEventRequest |
| from hive_metastore.ttypes import FireEventRequestData |
| from hive_metastore.ttypes import InsertEventRequestData |
| from tests.common.custom_cluster_test_suite import CustomClusterTestSuite |
| from tests.common.impala_test_suite import ImpalaTestSuite |
| from tests.common.skip import SkipIf, SkipIfFS |
| from tests.util.acid_txn import AcidTxn |
| from tests.util.hive_utils import HiveDbWrapper |
| from tests.util.event_processor_utils import EventProcessorUtils |
| from tests.util.filesystem_utils import WAREHOUSE |
| from tests.util.iceberg_util import IcebergCatalogs |
| |
| HIVE_SITE_HOUSEKEEPING_ON =\ |
| getenv('IMPALA_HOME') + '/fe/src/test/resources/hive-site-housekeeping-on' |
| |
| |
| @SkipIfFS.hive |
| class TestEventProcessingCustomConfigs(CustomClusterTestSuite): |
| """This class contains tests that exercise the event processing mechanism in the |
| catalog for non-default configurations""" |
| CATALOG_URL = "http://localhost:25020" |
| PROCESSING_TIMEOUT_S = 10 |
| |
| @pytest.mark.execute_serially |
| @CustomClusterTestSuite.with_args( |
| impalad_args="--blacklisted_dbs=testBlackListedDb " |
| "--blacklisted_tables=functional_parquet.testBlackListedTbl", |
| catalogd_args="--blacklisted_dbs=testBlackListedDb " |
| "--blacklisted_tables=functional_parquet.testBlackListedTbl " |
| "--hms_event_polling_interval_s=1") |
| def test_events_on_blacklisted_objects(self): |
| """Executes hive queries on blacklisted database and tables and makes sure that |
| event processor does not error out |
| """ |
| try: |
| event_id_before = EventProcessorUtils.get_last_synced_event_id() |
| # create a blacklisted database from hive and make sure event is ignored |
| self.run_stmt_in_hive("create database TESTblackListedDb") |
| # wait until all the events generated above are processed |
| EventProcessorUtils.wait_for_event_processing(self) |
| assert EventProcessorUtils.get_event_processor_status() == "ACTIVE" |
| assert EventProcessorUtils.get_last_synced_event_id() > event_id_before |
| # make sure that the blacklisted db is ignored |
| assert "TESTblackListedDb".lower() not in self.all_db_names() |
| |
| event_id_before = EventProcessorUtils.get_last_synced_event_id() |
| self.run_stmt_in_hive("create table testBlackListedDb.testtbl (id int)") |
| # create a table on the blacklisted database with a different case |
| self.run_stmt_in_hive("create table TESTBLACKlISTEDDb.t2 (id int)") |
| self.run_stmt_in_hive( |
| "create table functional_parquet.testBlackListedTbl (id int, val string)" |
| " partitioned by (part int) stored as parquet") |
| self.run_stmt_in_hive( |
| "alter table functional_parquet.testBlackListedTbl add partition (part=1)") |
| # wait until all the events generated above are processed |
| EventProcessorUtils.wait_for_event_processing(self) |
| assert EventProcessorUtils.get_event_processor_status() == "ACTIVE" |
| assert EventProcessorUtils.get_last_synced_event_id() > event_id_before |
| # make sure that the black listed table is not created |
| table_names = self.client.execute("show tables in functional_parquet").get_data() |
| assert "testBlackListedTbl".lower() not in table_names |
| |
| event_id_before = EventProcessorUtils.get_last_synced_event_id() |
| # generate a table level event with a different case |
| self.run_stmt_in_hive("drop table functional_parquet.TESTBlackListedTbl") |
| # wait until all the events generated above are processed |
| EventProcessorUtils.wait_for_event_processing(self) |
| assert EventProcessorUtils.get_event_processor_status() == "ACTIVE" |
| assert EventProcessorUtils.get_last_synced_event_id() > event_id_before |
| finally: |
| self.run_stmt_in_hive("drop database testBlackListedDb cascade") |
| self.run_stmt_in_hive("drop table functional_parquet.testBlackListedTbl") |
| |
| @CustomClusterTestSuite.with_args(catalogd_args="--hms_event_polling_interval_s=10") |
| def test_drop_table_events(self): |
| """IMPALA-10187: Event processing fails on multiple events + DROP TABLE. |
| This test issues ALTER TABLE + DROP in quick succession and checks whether event |
| processing still works. |
| """ |
| event_proc_timeout = 15 |
| db_name = ImpalaTestSuite.get_random_name("drop_event_db_") |
| with HiveDbWrapper(self, db_name): |
| tbl_name = "foo" |
| self.run_stmt_in_hive(""" |
| drop table if exists {db}.{tbl}; |
| create table {db}.{tbl} (id int); |
| insert into {db}.{tbl} values(1);""".format(db=db_name, tbl=tbl_name)) |
| # With MetastoreEventProcessor running, the insert event will be processed. Query |
| # the table from Impala. |
| EventProcessorUtils.wait_for_event_processing(self, event_proc_timeout) |
| # Verify that the data is present in Impala. |
| data = self.execute_scalar("select * from %s.%s" % (db_name, tbl_name)) |
| assert data == '1' |
| # Execute ALTER TABLE + DROP in quick succession so they will be processed in the |
| # same event batch. |
| self.run_stmt_in_hive(""" |
| alter table {db}.{tbl} set tblproperties ('foo'='bar'); |
| drop table {db}.{tbl};""".format(db=db_name, tbl=tbl_name)) |
| EventProcessorUtils.wait_for_event_processing(self, event_proc_timeout) |
| # Check that the event processor status is still ACTIVE. |
| assert EventProcessorUtils.get_event_processor_status() == "ACTIVE" |
| |
| @CustomClusterTestSuite.with_args( |
| cluster_size=1, |
| catalogd_args="--hms_event_polling_interval_s=1") |
| def test_create_drop_events(self, unique_database): |
| """Regression test for IMPALA-10502. The test runs very slow with default |
| statestored update frequency and hence this is changed to a custom cluster |
| test.""" |
| self.__run_create_drop_test(unique_database, "database") |
| self.__run_create_drop_test(unique_database, "table") |
| self.__run_create_drop_test(unique_database, "table", True) |
| self.__run_create_drop_test(unique_database, "table", True, True) |
| self.__run_create_drop_test(unique_database, "partition") |
| |
| @pytest.mark.execute_serially |
| @CustomClusterTestSuite.with_args( |
| impalad_args="--use_local_catalog=true", |
| catalogd_args="--catalog_topic_mode=minimal --hms_event_polling_interval_s=1", |
| cluster_size=1) |
| def test_local_catalog_create_drop_events(self, unique_database): |
| """ |
| Test is similar to the test_create_drop_events except this runs on local |
| """ |
| self.__run_create_drop_test(unique_database, "database") |
| self.__run_create_drop_test(unique_database, "table") |
| self.__run_create_drop_test(unique_database, "table", True) |
| self.__run_create_drop_test(unique_database, "table", True, True) |
| self.__run_create_drop_test(unique_database, "partition") |
| |
| def __run_create_drop_test(self, db, type, rename=False, rename_db=False): |
| if type == "table": |
| if not rename: |
| queries = [ |
| "create table {0}.test_{1} (i int)".format(db, 1), |
| "drop table {0}.test_{1}".format(db, 1) |
| ] |
| else: |
| db_1 = "{}_1".format(db) |
| if rename_db: |
| self.execute_query_expect_success(self.create_impala_client(), |
| "drop database if exists {0} cascade".format(db_1)) |
| self.execute_query_expect_success(self.create_impala_client(), |
| "create database {0}".format(db_1)) |
| self.execute_query_expect_success(self.create_impala_client(), |
| "create table if not exists {0}.rename_test_1 (i int)".format(db)) |
| if rename_db: |
| queries = [ |
| "alter table {0}.rename_test_1 rename to {1}.rename_test_1".format(db, |
| db_1), |
| "alter table {0}.rename_test_1 rename to {1}.rename_test_1".format(db_1, db) |
| ] |
| else: |
| queries = [ |
| "alter table {0}.rename_test_1 rename to {0}.rename_test_2".format(db), |
| "alter table {0}.rename_test_2 rename to {0}.rename_test_1".format(db) |
| ] |
| create_metric_name = "tables-added" |
| removed_metric_name = "tables-removed" |
| elif type == "database": |
| self.execute_query_expect_success(self.create_impala_client(), |
| "drop database if exists {0}".format("test_create_drop_db")) |
| queries = [ |
| "create database {db}".format(db="test_create_drop_db"), |
| "drop database {db}".format(db="test_create_drop_db") |
| ] |
| create_metric_name = "databases-added" |
| removed_metric_name = "databases-removed" |
| else: |
| tbl_name = "test_create_drop_partition" |
| self.execute_query_expect_success(self.create_impala_client(), |
| "create table {db}.{tbl} (c int) partitioned by (p int)".format( |
| db=db, tbl=tbl_name)) |
| queries = [ |
| "alter table {db}.{tbl} add partition (p=1)".format(db=db, tbl=tbl_name), |
| "alter table {db}.{tbl} drop partition (p=1)".format(db=db, tbl=tbl_name) |
| ] |
| create_metric_name = "partitions-added" |
| removed_metric_name = "partitions-removed" |
| |
| # get the metric before values |
| EventProcessorUtils.wait_for_event_processing(self) |
| create_metric_val_before = EventProcessorUtils.get_int_metric(create_metric_name, 0) |
| removed_metric_val_before = EventProcessorUtils.get_int_metric(removed_metric_name, 0) |
| events_skipped_before = EventProcessorUtils.get_int_metric('events-skipped', 0) |
| num_iters = 100 |
| for iter in range(num_iters): |
| for q in queries: |
| try: |
| self.execute_query_expect_success(self.create_impala_client(), q) |
| except Exception as e: |
| print("Failed in {} iterations. Error {}".format(iter, str(e))) |
| raise |
| EventProcessorUtils.wait_for_event_processing(self) |
| create_metric_val_after = EventProcessorUtils.get_int_metric(create_metric_name, 0) |
| removed_metric_val_after = EventProcessorUtils.get_int_metric(removed_metric_name, 0) |
| events_skipped_after = EventProcessorUtils.get_int_metric('events-skipped', 0) |
| num_delete_event_entries = EventProcessorUtils.\ |
| get_int_metric('delete-event-log-size', 0) |
| assert EventProcessorUtils.get_event_processor_status() == "ACTIVE" |
| # None of the queries above should actually trigger a add/remove object from events |
| assert create_metric_val_after == create_metric_val_before |
| assert removed_metric_val_after == removed_metric_val_before |
| # each query set generates 2 events and both of them should be skipped |
| assert events_skipped_after == num_iters * 2 + events_skipped_before |
| # make sure that there are no more entries in the delete event log |
| assert num_delete_event_entries == 0 |
| |
| @CustomClusterTestSuite.with_args(catalogd_args="--hms_event_polling_interval_s=1") |
| def test_self_events(self, unique_database): |
| """Runs multiple queries which generate events and makes |
| sure that tables and partitions are not refreshed the queries is run from Impala. If |
| the queries are run from Hive, we make sure that the tables and partitions are |
| refreshed""" |
| self.__run_self_events_test(unique_database, True) |
| self.__run_self_events_test(unique_database, False) |
| |
| @CustomClusterTestSuite.with_args( |
| catalogd_args="--hms_event_polling_interval_s=5" |
| " --enable_reload_events=true") |
| def test_refresh_invalidate_events(self, unique_database): |
| self.run_test_refresh_invalidate_events(unique_database, "reload_table") |
| |
| @CustomClusterTestSuite.with_args( |
| catalogd_args="--hms_event_polling_interval_s=5" |
| " --enable_reload_events=true" |
| " --enable_sync_to_latest_event_on_ddls=true") |
| def test_refresh_invalidate_events_enable_sync_to_latest_events(self, unique_database): |
| self.run_test_refresh_invalidate_events(unique_database, "reload_table_sync", True) |
| |
| def run_test_refresh_invalidate_events(self, unique_database, test_reload_table, |
| enable_sync_to_latest_event_on_ddls=False): |
| """Test is to verify Impala-11808, refresh/invalidate commands should generate a |
| Reload event in HMS and CatalogD's event processor should process this event. |
| """ |
| self.client.execute( |
| "create table {}.{} (i int) partitioned by (year int) " |
| .format(unique_database, test_reload_table)) |
| self.client.execute( |
| "insert into {}.{} partition (year=2022) values (1),(2),(3)" |
| .format(unique_database, test_reload_table)) |
| self.client.execute( |
| "insert into {}.{} partition (year=2023) values (1),(2),(3)" |
| .format(unique_database, test_reload_table)) |
| EventProcessorUtils.wait_for_event_processing(self) |
| |
| def check_self_events(query): |
| tbls_refreshed_before, partitions_refreshed_before, \ |
| events_skipped_before = self.__get_self_event_metrics() |
| last_event_id = EventProcessorUtils.get_current_notification_id(self.hive_client) |
| self.client.execute(query) |
| # Check if there is a reload event fired after refresh query. |
| events = EventProcessorUtils.get_next_notification(self.hive_client, last_event_id) |
| assert len(events) == 1 |
| last_event = events[0] |
| assert last_event.dbName == unique_database |
| assert last_event.tableName == test_reload_table |
| assert last_event.eventType == "RELOAD" |
| EventProcessorUtils.wait_for_event_processing(self) |
| tbls_refreshed_after, partitions_refreshed_after, \ |
| events_skipped_after = self.__get_self_event_metrics() |
| assert events_skipped_after > events_skipped_before |
| |
| check_self_events("refresh {}.{} partition(year=2022)" |
| .format(unique_database, test_reload_table)) |
| check_self_events("refresh {}.{}".format(unique_database, test_reload_table)) |
| EventProcessorUtils.wait_for_event_processing(self) |
| |
| if enable_sync_to_latest_event_on_ddls: |
| # Test to verify if older events are being skipped in event processor |
| data = FireEventRequestData() |
| data.refreshEvent = True |
| req = FireEventRequest(True, data) |
| req.dbName = unique_database |
| req.tableName = test_reload_table |
| # table level reload events |
| tbl_events_skipped_before = EventProcessorUtils.get_num_skipped_events() |
| for i in range(10): |
| self.hive_client.fire_listener_event(req) |
| EventProcessorUtils.wait_for_event_processing(self) |
| tbl_events_skipped_after = EventProcessorUtils.get_num_skipped_events() |
| assert tbl_events_skipped_after > tbl_events_skipped_before |
| # partition level reload events |
| EventProcessorUtils.wait_for_event_processing(self) |
| part_events_skipped_before = EventProcessorUtils.get_num_skipped_events() |
| req.partitionVals = ["2022"] |
| for i in range(10): |
| self.hive_client.fire_listener_event(req) |
| EventProcessorUtils.wait_for_event_processing(self) |
| part_events_skipped_after = EventProcessorUtils.get_num_skipped_events() |
| assert part_events_skipped_after > part_events_skipped_before |
| |
| # Test to verify IMPALA-12213 |
| table = self.hive_client.get_table(unique_database, test_reload_table) |
| table.dbName = unique_database |
| table.tableName = "test_sequence_table" |
| self.hive_client.create_table(table) |
| data = FireEventRequestData() |
| data.refreshEvent = True |
| req = FireEventRequest(True, data) |
| req.dbName = unique_database |
| req.tableName = "test_sequence_table" |
| self.hive_client.fire_listener_event(req) |
| EventProcessorUtils.wait_for_event_processing(self) |
| assert EventProcessorUtils.get_event_processor_status() == "ACTIVE" |
| |
| @CustomClusterTestSuite.with_args( |
| catalogd_args="--enable_reload_events=true") |
| def test_reload_events_with_transient_partitions(self, unique_database): |
| tbl = unique_database + ".tbl" |
| create_stmt = "create table {} (i int) partitioned by(p int)".format(tbl) |
| add_part_stmt = "alter table {} add if not exists partition(p=0)".format(tbl) |
| drop_part_stmt = "alter table {} drop if exists partition(p=0)".format(tbl) |
| refresh_stmt = "refresh {} partition(p=0)".format(tbl) |
| end_states = [self.client.QUERY_STATES['FINISHED'], |
| self.client.QUERY_STATES['EXCEPTION']] |
| |
| self.execute_query(create_stmt) |
| self.execute_query(add_part_stmt) |
| # Run REFRESH partition in the background so we can drop the partition concurrently. |
| refresh_handle = self.client.execute_async(refresh_stmt) |
| # Before IMPALA-12855, REFRESH usually fails in 2-3 rounds. |
| for i in range(100): |
| self.execute_query(drop_part_stmt) |
| refresh_state = self.wait_for_any_state(refresh_handle, end_states, 10) |
| assert refresh_state == self.client.QUERY_STATES['FINISHED'],\ |
| "REFRESH state: {}. Error log: {}".format( |
| QueryState._VALUES_TO_NAMES[refresh_state], |
| self.client.get_log(refresh_handle)) |
| self.execute_query(add_part_stmt) |
| refresh_handle = self.client.execute_async(refresh_stmt) |
| |
| @CustomClusterTestSuite.with_args( |
| catalogd_args="--hms_event_polling_interval_s=10" |
| " --enable_skipping_older_events=true" |
| " --enable_sync_to_latest_event_on_ddls=true") |
| def test_skipping_older_events(self, unique_database): |
| """Test is to verify IMPALA-11535, event processor should ignore older events if the |
| current event id is older than the lastRefreshEventId on the table/partition |
| """ |
| test_old_table = "test_old_table" |
| |
| def verify_skipping_older_events(table_name, is_transactional, is_partitioned): |
| query = " ".join(["create", "transactional" if is_transactional else '', |
| "table {}.{} (i int)", "partitioned by (year int)" if is_partitioned else '']) |
| self.run_stmt_in_hive(query.format(unique_database, table_name)) |
| values = "values (10),(20),(30)" |
| EventProcessorUtils.wait_for_event_processing(self) |
| |
| def verify_skipping_hive_stmt_events(stmt, new_table_name): |
| tbl_events_skipped_before = EventProcessorUtils.get_num_skipped_events() |
| self.run_stmt_in_hive(stmt) |
| self.client.execute( |
| "refresh {}.{}".format(unique_database, new_table_name)) |
| tables_refreshed_before = EventProcessorUtils.get_int_metric("tables-refreshed") |
| partitions_refreshed_before = \ |
| EventProcessorUtils.get_int_metric("partitions-refreshed") |
| EventProcessorUtils.wait_for_event_processing(self) |
| tbl_events_skipped_after = EventProcessorUtils.get_num_skipped_events() |
| assert tbl_events_skipped_after > tbl_events_skipped_before |
| tables_refreshed_after = EventProcessorUtils.get_int_metric("tables-refreshed") |
| partitions_refreshed_after = \ |
| EventProcessorUtils.get_int_metric("partitions-refreshed") |
| if is_partitioned: |
| assert partitions_refreshed_after == partitions_refreshed_before |
| else: |
| assert tables_refreshed_after == tables_refreshed_before |
| |
| # test single insert event |
| query = " ".join(["insert into `{}`.`{}`", "partition (year=2023)" |
| if is_partitioned else '', values]) |
| verify_skipping_hive_stmt_events( |
| query.format(unique_database, table_name), table_name) |
| # test batch insert events |
| query = " ".join(["insert into `{}`.`{}`", "partition (year=2023)" |
| if is_partitioned else '', values, ";"]) |
| complete_query = "" |
| for _ in range(3): |
| complete_query += query.format(unique_database, table_name) |
| verify_skipping_hive_stmt_events(complete_query, table_name) |
| # Dynamic partitions test |
| query = " ".join(["create", "table `{}`.`{}` (i int)", |
| " partitioned by (year int) " if is_partitioned else '', |
| self.__get_transactional_tblproperties(is_transactional)]) |
| self.client.execute(query.format(unique_database, "new_table")) |
| complete_query = "insert overwrite table `{db}`.`{tbl1}` " \ |
| "select * from `{db}`.`{tbl2}`"\ |
| .format(db=unique_database, tbl1="new_table", tbl2=table_name) |
| verify_skipping_hive_stmt_events(complete_query, "new_table") |
| # Drop the tables before running another test |
| self.client.execute("drop table {}.{}".format(unique_database, table_name)) |
| self.client.execute("drop table {}.{}".format(unique_database, "new_table")) |
| verify_skipping_older_events(test_old_table, False, False) |
| verify_skipping_older_events(test_old_table, True, False) |
| verify_skipping_older_events(test_old_table, False, True) |
| verify_skipping_older_events(test_old_table, True, True) |
| |
| @CustomClusterTestSuite.with_args( |
| catalogd_args="--enable_sync_to_latest_event_on_ddls=true " |
| "--debug_actions=catalogd_get_filtered_events_delay:SLEEP@3000 ") |
| def test_skipping_batching_events(self, unique_database): |
| """Test to verify IMPALA-10949, improving batching logic for partition events. |
| Before batching the events, each event is checked if the event id is greater than |
| table's lastSyncEventId then the event can be batched else it can be skipped.""" |
| # Print trace logs from DebugUtils. |
| self.cluster.catalogd.set_jvm_log_level("org.apache.impala.util.DebugUtils", "trace") |
| test_batch_table = "test_batch_table" |
| self.client.execute( |
| "create table {}.{} like functional.alltypes" |
| .format(unique_database, test_batch_table)) |
| self.client.execute( |
| "insert into {}.{} partition (year,month) select * from functional.alltypes" |
| .format(unique_database, test_batch_table)) |
| # Generate batch ALTER_PARTITION events |
| self.run_stmt_in_hive( |
| "analyze table {}.{} compute statistics".format(unique_database, test_batch_table)) |
| EventProcessorUtils.wait_for_event_processing(self) |
| batch_events_metric = "batch-events-created" |
| batch_events_1 = EventProcessorUtils.get_int_metric(batch_events_metric) |
| prev_skipped_events = EventProcessorUtils.get_int_metric("events-skipped") |
| self.run_stmt_in_hive( |
| "analyze table {}.{} compute statistics".format(unique_database, test_batch_table)) |
| self.client.execute("refresh {0}.{1}".format(unique_database, test_batch_table)) |
| EventProcessorUtils.wait_for_event_processing(self) |
| batch_events_2 = EventProcessorUtils.get_int_metric(batch_events_metric) |
| current_skipped_events = EventProcessorUtils.get_int_metric("events-skipped") |
| # Make sure no new batch events are created |
| assert batch_events_2 == batch_events_1 |
| assert current_skipped_events - prev_skipped_events >= 24 |
| |
| @CustomClusterTestSuite.with_args(catalogd_args="--hms_event_polling_interval_s=1") |
| def test_commit_compaction_events(self, unique_database): |
| """Test is to verify Impala-11626, commit compaction events triggered in HMS would |
| be consumed by CatalogD's event processor. |
| """ |
| |
| # Test scenario 1: partitioned table |
| test_cc_part_table = "test_cc_partitioned_table" |
| self.run_stmt_in_hive( |
| "create transactional table {}.{} (i int) partitioned by (year int)" |
| .format(unique_database, test_cc_part_table)) |
| for i in range(2): |
| self.run_stmt_in_hive( |
| "insert into {}.{} partition (year=2022) values (1),(2),(3)" |
| .format(unique_database, test_cc_part_table)) |
| EventProcessorUtils.wait_for_event_processing(self) |
| parts_refreshed_before_compaction = EventProcessorUtils.get_int_metric( |
| "partitions-refreshed") |
| self.client.execute( |
| "select * from {}.{} limit 2" |
| .format(unique_database, test_cc_part_table)) |
| self.run_stmt_in_hive( |
| "alter table {}.{} partition(year=2022) compact 'minor' and wait" |
| .format(unique_database, test_cc_part_table)) |
| EventProcessorUtils.wait_for_event_processing(self) |
| parts_refreshed_after_compaction = EventProcessorUtils.get_int_metric( |
| "partitions-refreshed") |
| assert parts_refreshed_after_compaction > parts_refreshed_before_compaction |
| |
| # Test scenario 2: |
| test_cc_unpart_tab = "test_cc_unpart_table" |
| self.run_stmt_in_hive( |
| "create transactional table {}.{} (i int)" |
| .format(unique_database, test_cc_unpart_tab)) |
| for i in range(2): |
| self.run_stmt_in_hive( |
| "insert into {}.{} values (1),(2),(3)" |
| .format(unique_database, test_cc_unpart_tab)) |
| EventProcessorUtils.wait_for_event_processing(self) |
| tables_refreshed_before_compaction = EventProcessorUtils.get_int_metric( |
| "tables-refreshed") |
| self.client.execute( |
| "select * from {}.{} limit 2" |
| .format(unique_database, test_cc_unpart_tab)) |
| self.run_stmt_in_hive("alter table {}.{} compact 'minor' and wait" |
| .format(unique_database, test_cc_unpart_tab)) |
| EventProcessorUtils.wait_for_event_processing(self) |
| tables_refreshed_after_compaction = EventProcessorUtils.get_int_metric( |
| "tables-refreshed") |
| assert tables_refreshed_after_compaction > tables_refreshed_before_compaction |
| |
| # Test scenario 3: partitioned table has partition deleted |
| test_cc_part_table = "test_cc_partitioned_table_error" |
| self.run_stmt_in_hive( |
| "create transactional table {}.{} (i int) partitioned by (year int)" |
| .format(unique_database, test_cc_part_table)) |
| for i in range(2): |
| self.run_stmt_in_hive( |
| "insert into {}.{} partition (year=2022) values (1),(2),(3)" |
| .format(unique_database, test_cc_part_table)) |
| EventProcessorUtils.wait_for_event_processing(self) |
| self.client.execute( |
| "select * from {}.{} limit 2" |
| .format(unique_database, test_cc_part_table)) |
| self.run_stmt_in_hive( |
| "alter table {}.{} partition(year=2022) compact 'minor' and wait" |
| .format(unique_database, test_cc_part_table)) |
| self.run_stmt_in_hive("alter table {}.{} Drop if exists partition(year=2022)" |
| .format(unique_database, test_cc_part_table)) |
| EventProcessorUtils.wait_for_event_processing(self) |
| assert EventProcessorUtils.get_event_processor_status() == "ACTIVE" |
| |
| # Test scenario 4: process commit compaction for an unloaded table |
| test_cc_part_table = "test_cc_table_unloaded" |
| self.run_stmt_in_hive( |
| "create transactional table {}.{} (i int) partitioned by (year int)" |
| .format(unique_database, test_cc_part_table)) |
| for i in range(2): |
| self.run_stmt_in_hive( |
| "insert into {}.{} partition (year=2022) values (1),(2),(3)" |
| .format(unique_database, test_cc_part_table)) |
| EventProcessorUtils.wait_for_event_processing(self) |
| self.run_stmt_in_hive( |
| "alter table {}.{} partition(year=2022) compact 'minor' and wait" |
| .format(unique_database, test_cc_part_table)) |
| EventProcessorUtils.wait_for_event_processing(self) |
| assert EventProcessorUtils.get_event_processor_status() == "ACTIVE" |
| |
| @CustomClusterTestSuite.with_args(catalogd_args="--hms_event_polling_interval_s=1") |
| def test_event_batching(self, unique_database): |
| """Runs queries which generate multiple ALTER_PARTITION events which must be |
| batched by events processor. Runs as a custom cluster test to isolate the metric |
| values from other tests.""" |
| testtbl = "test_event_batching" |
| test_acid_tbl = "test_event_batching_acid" |
| acid_props = self.__get_transactional_tblproperties(True) |
| # create test tables |
| self.client.execute( |
| "create table {}.{} like functional.alltypes".format(unique_database, testtbl)) |
| self.client.execute( |
| "insert into {}.{} partition (year,month) select * from functional.alltypes".format( |
| unique_database, testtbl)) |
| self.client.execute( |
| "create table {}.{} (id int) partitioned by (year int, month int) {}".format( |
| unique_database, test_acid_tbl, acid_props)) |
| self.client.execute( |
| "insert into {}.{} partition (year, month) " |
| "select id, year, month from functional.alltypes".format(unique_database, |
| test_acid_tbl)) |
| # run compute stats from impala; this should generate 24 ALTER_PARTITION events which |
| # should be batched together into 1 or more number of events. |
| EventProcessorUtils.wait_for_event_processing(self) |
| batch_events_metric = "batch-events-created" |
| batch_events_1 = EventProcessorUtils.get_int_metric(batch_events_metric) |
| self.client.execute("compute stats {}.{}".format(unique_database, testtbl)) |
| EventProcessorUtils.wait_for_event_processing(self) |
| batch_events_2 = EventProcessorUtils.get_int_metric(batch_events_metric) |
| assert batch_events_2 > batch_events_1 |
| # run analyze stats event from hive which generates ALTER_PARTITION event on each |
| # partition of the table |
| self.run_stmt_in_hive( |
| "analyze table {}.{} compute statistics".format(unique_database, testtbl)) |
| EventProcessorUtils.wait_for_event_processing(self) |
| batch_events_3 = EventProcessorUtils.get_int_metric(batch_events_metric) |
| assert batch_events_3 > batch_events_2 |
| # in case of transactional table since we batch the events together, the number of |
| # tables refreshed must be far lower than number of events generated |
| num_table_refreshes_1 = EventProcessorUtils.get_int_metric( |
| "tables-refreshed") |
| self.client.execute("compute stats {}.{}".format(unique_database, test_acid_tbl)) |
| EventProcessorUtils.wait_for_event_processing(self) |
| batch_events_4 = EventProcessorUtils.get_int_metric(batch_events_metric) |
| num_table_refreshes_2 = EventProcessorUtils.get_int_metric( |
| "tables-refreshed") |
| # we should generate atleast 1 batch event if not more due to the 24 consecutive |
| # ALTER_PARTITION events |
| assert batch_events_4 > batch_events_3 |
| # table should not be refreshed since this is a self-event |
| assert num_table_refreshes_2 == num_table_refreshes_1 |
| self.run_stmt_in_hive( |
| "analyze table {}.{} compute statistics".format(unique_database, test_acid_tbl)) |
| EventProcessorUtils.wait_for_event_processing(self) |
| batch_events_5 = EventProcessorUtils.get_int_metric(batch_events_metric) |
| assert batch_events_5 > batch_events_4 |
| num_table_refreshes_2 = EventProcessorUtils.get_int_metric("tables-refreshed") |
| # the analyze table from hive generates 24 ALTER_PARTITION events which should be |
| # batched into 1-2 batches (depending on timing of the event poll thread). |
| assert num_table_refreshes_2 > num_table_refreshes_1 |
| assert int(num_table_refreshes_2) - int(num_table_refreshes_1) < 24 |
| EventProcessorUtils.wait_for_event_processing(self) |
| # test for batching of insert events |
| batch_events_insert = EventProcessorUtils.get_int_metric(batch_events_metric) |
| tables_refreshed_insert = EventProcessorUtils.get_int_metric("tables-refreshed") |
| partitions_refreshed_insert = EventProcessorUtils.get_int_metric( |
| "partitions-refreshed") |
| self.client.execute( |
| "insert into {}.{} partition (year,month) select * from functional.alltypes".format( |
| unique_database, testtbl)) |
| EventProcessorUtils.wait_for_event_processing(self) |
| batch_events_after_insert = EventProcessorUtils.get_int_metric(batch_events_metric) |
| tables_refreshed_after_insert = EventProcessorUtils.get_int_metric("tables-refreshed") |
| partitions_refreshed_after_insert = EventProcessorUtils.get_int_metric( |
| "partitions-refreshed") |
| # this is a self-event tables or partitions should not be refreshed |
| assert batch_events_after_insert > batch_events_insert |
| assert tables_refreshed_after_insert == tables_refreshed_insert |
| assert partitions_refreshed_after_insert == partitions_refreshed_insert |
| # run the insert from hive to make sure that batch event is refreshing all the |
| # partitions |
| self.run_stmt_in_hive( |
| "SET hive.exec.dynamic.partition.mode=nonstrict; insert into {}.{} partition" |
| " (year,month) select * from functional.alltypes".format( |
| unique_database, testtbl)) |
| EventProcessorUtils.wait_for_event_processing(self) |
| batch_events_after_hive = EventProcessorUtils.get_int_metric(batch_events_metric) |
| partitions_refreshed_after_hive = EventProcessorUtils.get_int_metric( |
| "partitions-refreshed") |
| assert batch_events_after_hive > batch_events_insert |
| # 24 partitions inserted and hence we must refresh 24 partitions once. |
| assert int(partitions_refreshed_after_hive) == int(partitions_refreshed_insert) + 24 |
| |
| @CustomClusterTestSuite.with_args(catalogd_args="--hms_event_polling_interval_s=5") |
| def test_event_processor_failure_extra_space(self, unique_database): |
| """This test verifies that impala event processor is in active state after |
| processing a couple of previously erroneous events""" |
| test_table = "extra_space_table" |
| # IMPALA-11939 -- create table event in HMS contains extra spaces in the db/table |
| self.run_stmt_in_hive("create table ` {}`.`{} ` (i1 int) partitioned by (year int)" |
| .format(unique_database, test_table)) |
| self.run_stmt_in_hive("alter table ` {}`.`{} ` add columns (i2 int)" |
| .format(unique_database, test_table)) |
| EventProcessorUtils.wait_for_event_processing(self) |
| assert EventProcessorUtils.get_event_processor_status() == "ACTIVE" |
| |
| @CustomClusterTestSuite.with_args(catalogd_args="--hms_event_polling_interval_s=5") |
| def test_disable_hms_sync(self, unique_database): |
| """This test verifies that impala event processor is in active state after |
| processing an alter table event that re-enables hms sync""" |
| # test 1: re-enable disableHmsSync config at table level |
| test_table = "disable_hms_sync_table" |
| self.client.execute( |
| """create table {}.{} (i int) TBLPROPERTIES ('impala.disableHmsSync'='true')""" |
| .format(unique_database, test_table)) |
| EventProcessorUtils.wait_for_event_processing(self) |
| prev_events_skipped = EventProcessorUtils.get_int_metric('events-skipped', 0) |
| self.run_stmt_in_hive( |
| """ALTER TABLE {}.{} SET TBLPROPERTIES('somekey'='somevalue')""" |
| .format(unique_database, test_table)) |
| self.client.execute( |
| """ALTER TABLE {}.{} SET TBLPROPERTIES ('impala.disableHmsSync'='false')""" |
| .format(unique_database, test_table)) |
| EventProcessorUtils.wait_for_event_processing(self) |
| current_events_skipped = EventProcessorUtils.get_int_metric('events-skipped', 0) |
| assert EventProcessorUtils.get_event_processor_status() == "ACTIVE" |
| assert current_events_skipped >= prev_events_skipped + 1 |
| |
| # test 2: re-enabling disableHmsSync config on a table shouldn't put event processor |
| # in error state if the database is not loaded. |
| try: |
| test_db = "unloaded_db_sync" |
| self.run_stmt_in_hive("""create database {}""".format(test_db)) |
| self.run_stmt_in_hive("""create table {}.{} (id int) |
| TBLPROPERTIES ('impala.disableHmsSync'='true')""".format(test_db, test_table)) |
| self.run_stmt_in_hive( |
| """ALTER TABLE {}.{} SET TBLPROPERTIES ('impala.disableHmsSync'='false')""" |
| .format(test_db, test_table)) |
| EventProcessorUtils.wait_for_event_processing(self) |
| assert EventProcessorUtils.get_event_processor_status() == "ACTIVE" |
| finally: |
| self.run_stmt_in_hive("""drop database {} cascade""".format(test_db)) |
| |
| @CustomClusterTestSuite.with_args(catalogd_args="--hms_event_polling_interval_s=10") |
| def test_event_processor_dropped_partition(self, unique_database): |
| """This test verifies that impala event processor is in active state after |
| processing partitioned insert events of a dropped table""" |
| # IMPALA-11768 -- Insert partition events should be ignored |
| # if the table is dropped |
| test_table = "partitioned_table" |
| |
| def is_event_processor_active(is_insert): |
| self.run_stmt_in_hive("create table {}.{} (i1 int) partitioned by (year int)" |
| .format(unique_database, test_table)) |
| EventProcessorUtils.wait_for_event_processing(self) |
| self.client.execute("refresh {}.{}".format(unique_database, test_table)) |
| self.run_stmt_in_hive( |
| "insert into {}.{} partition(year=2023) values (4),(5),(6)" |
| .format(unique_database, test_table)) |
| data = FireEventRequestData() |
| if is_insert: |
| insert_data = InsertEventRequestData() |
| insert_data.filesAdded = "/warehouse/mytable/b1" |
| insert_data.replace = False |
| data.insertData = insert_data |
| else: |
| data.refreshEvent = True |
| req = FireEventRequest(True, data) |
| req.dbName = unique_database |
| req.tableName = test_table |
| req.partitionVals = ["2023"] |
| self.hive_client.fire_listener_event(req) |
| self.run_stmt_in_hive( |
| "drop table {}.{}".format(unique_database, test_table)) |
| EventProcessorUtils.wait_for_event_processing(self) |
| assert EventProcessorUtils.get_event_processor_status() == "ACTIVE" |
| |
| is_event_processor_active(True) |
| is_event_processor_active(False) |
| |
| @CustomClusterTestSuite.with_args(catalogd_args="--hms_event_polling_interval_s=1") |
| def test_iceberg_self_events(self, unique_database): |
| """This test checks that Impala doesn't refresh Iceberg tables on self events.""" |
| tbl_name = unique_database + ".test_iceberg_events" |
| iceberg_catalogs = IcebergCatalogs(unique_database) |
| |
| def check_self_events(query, skips_events=True): |
| tbls_refreshed_before, partitions_refreshed_before, \ |
| events_skipped_before = self.__get_self_event_metrics() |
| self.client.execute(query) |
| EventProcessorUtils.wait_for_event_processing(self) |
| tbls_refreshed_after, partitions_refreshed_after, \ |
| events_skipped_after = self.__get_self_event_metrics() |
| assert tbls_refreshed_before == tbls_refreshed_after |
| assert partitions_refreshed_before == partitions_refreshed_after |
| if skips_events: |
| assert events_skipped_after > events_skipped_before |
| |
| for catalog in iceberg_catalogs.get_iceberg_catalog_properties(): |
| is_hive_catalog = iceberg_catalogs.is_a_hive_catalog(catalog) |
| self.client.execute(""" |
| CREATE TABLE {0} (i int) STORED AS ICEBERG |
| TBLPROPERTIES ({1})""".format(tbl_name, catalog)) |
| |
| check_self_events("INSERT OVERWRITE {0} VALUES (1)".format(tbl_name), |
| skips_events=is_hive_catalog) |
| check_self_events("ALTER TABLE {0} ADD COLUMN j INT".format(tbl_name)) |
| check_self_events("ALTER TABLE {0} DROP COLUMN i".format(tbl_name)) |
| check_self_events("ALTER TABLE {0} CHANGE COLUMN j j BIGINT".format(tbl_name)) |
| # SET PARTITION SPEC only updates HMS in case of HiveCatalog (which sets |
| # table property 'metadata_location') |
| check_self_events( |
| "ALTER TABLE {0} SET PARTITION SPEC (truncate(2, j))".format(tbl_name), |
| skips_events=is_hive_catalog) |
| check_self_events( |
| "ALTER TABLE {0} SET TBLPROPERTIES('key'='value')".format(tbl_name)) |
| check_self_events("ALTER TABLE {0} UNSET TBLPROPERTIES('key')".format(tbl_name)) |
| check_self_events("INSERT INTO {0} VALUES (2), (3), (4)".format(tbl_name), |
| skips_events=is_hive_catalog) |
| ctas_tbl = unique_database + ".ice_ctas" |
| check_self_events("""CREATE TABLE {0} STORED AS ICEBERG |
| TBLPROPERTIES ({1}) AS SELECT * FROM {2}""".format(ctas_tbl, catalog, tbl_name)) |
| check_self_events("DROP TABLE {0}".format(ctas_tbl)) |
| check_self_events("TRUNCATE TABLE {0}".format(tbl_name), |
| skips_events=is_hive_catalog) |
| |
| self.client.execute("DROP TABLE {0}".format(tbl_name)) |
| |
| def __run_self_events_test(self, db_name, use_impala): |
| recover_tbl_name = ImpalaTestSuite.get_random_name("tbl_") |
| # create a table similar to alltypes so that we can recover the partitions on it |
| # later in one of the test queries |
| alltypes_tab_location = self.__get_tbl_location("functional", "alltypes") |
| self.client.execute( |
| "create external table {0}.{1} like functional.alltypes location '{2}'".format( |
| db_name, recover_tbl_name, alltypes_tab_location)) |
| self.client.execute("refresh {0}.{1}".format(db_name, recover_tbl_name)) |
| if use_impala: |
| queries = self.__get_impala_test_queries(db_name, recover_tbl_name) |
| # some queries do not trigger self-event evaluation (creates and drops) however, |
| # its still good to confirm that we don't do unnecessary refreshes in such cases |
| # For such queries we use a different metrics events-skipped to confirm that these |
| # events are skipped. |
| for stmt in queries[False]: |
| self.__exec_sql_and_check_selfevent_counter(stmt, use_impala, False) |
| # All the queries with True key should confirm that the events-skipped counter |
| # is also incremented |
| for stmt in queries[True]: |
| self.__exec_sql_and_check_selfevent_counter(stmt, use_impala) |
| else: |
| queries = self.__get_hive_test_queries(db_name, recover_tbl_name) |
| for stmt in queries: |
| self.__exec_sql_and_check_selfevent_counter(stmt, use_impala) |
| |
| def __get_impala_test_queries(self, db_name, recover_tbl_name): |
| tbl_name = ImpalaTestSuite.get_random_name("tbl_") |
| acid_tbl_name = ImpalaTestSuite.get_random_name("acid_tbl_") |
| acid_no_part_tbl_name = ImpalaTestSuite.get_random_name("acid_no_part_tbl_") |
| tbl2 = ImpalaTestSuite.get_random_name("tbl_") |
| view_name = ImpalaTestSuite.get_random_name("view_") |
| view2 = ImpalaTestSuite.get_random_name("view_") |
| # create a empty table for both partitioned and unpartitioned case for testing insert |
| # events |
| empty_unpartitioned_tbl = ImpalaTestSuite.get_random_name("empty_unpart_tbl_") |
| empty_partitioned_tbl = ImpalaTestSuite.get_random_name("empty_parttbl_") |
| self.client.execute( |
| "create table {0}.{1} (c1 int)".format(db_name, empty_unpartitioned_tbl)) |
| self.client.execute( |
| "create table {0}.{1} (c1 int) partitioned by (part int)".format(db_name, |
| empty_partitioned_tbl)) |
| acid_props = self.__get_transactional_tblproperties(True) |
| self_event_test_queries = { |
| # Queries which will increment the events-skipped counter |
| True: [ |
| # ALTER_DATABASE case |
| "comment on database {0} is 'self-event test database'".format(db_name), |
| "alter database {0} set owner user `test-user`".format(db_name), |
| "create function {0}.f() returns int location '{1}/libTestUdfs.so' " |
| "symbol='NoArgs'".format(db_name, WAREHOUSE), |
| "drop function {0}.f()".format(db_name), |
| # ALTER_TABLE case |
| "alter table {0}.{1} set TBLPROPERTIES ('k'='v')".format(db_name, tbl_name), |
| "alter table {0}.{1} ADD COLUMN c1 int".format(db_name, tbl_name), |
| "alter table {0}.{1} ALTER COLUMN C1 set comment 'c1 comment'".format(db_name, |
| tbl_name), |
| "comment on table {0}.{1} IS 'table level comment'".format(db_name, tbl_name), |
| "comment on column {0}.{1}.C1 IS 'column level comment'".format(db_name, |
| tbl_name), |
| "alter table {0}.{1} ADD COLUMNS (c2 int, c3 string)".format(db_name, tbl_name), |
| "alter table {0}.{1} DROP COLUMN c1".format(db_name, tbl_name), |
| "alter table {0}.{1} DROP COLUMN c2".format(db_name, tbl_name), |
| "alter table {0}.{1} DROP COLUMN c3".format(db_name, tbl_name), |
| "alter table {0}.{1} set owner user `test-user`".format(db_name, tbl_name), |
| "alter table {0}.{1} set owner role `test-role`".format(db_name, tbl_name), |
| "alter view {0}.{1} set owner user `test-view-user`".format(db_name, view_name), |
| "alter view {0}.{1} set owner role `test-view-role`".format(db_name, view_name), |
| # compute stats will generates ALTER_PARTITION |
| "compute stats {0}.{1}".format(db_name, tbl_name), |
| "compute incremental stats {0}.{1}".format(db_name, tbl_name), |
| "drop stats {0}.{1}".format(db_name, tbl_name), |
| # insert into a existing partition; generates INSERT self-event |
| "insert into table {0}.{1} partition " |
| "(year, month) select * from functional.alltypessmall where year=2009 " |
| "and month=1".format(db_name, tbl_name), |
| # insert overwrite query from Impala also generates a INSERT self-event |
| "insert overwrite table {0}.{1} partition " |
| "(year, month) select * from functional.alltypessmall where year=2009 " |
| "and month=1".format(db_name, tbl_name), |
| # events processor doesn't process delete column stats events currently, |
| # however, in case of incremental stats, there could be alter table and |
| # alter partition events which should be ignored. Hence we run compute stats |
| # before to make sure that the truncate table command generated alter events |
| # are ignored. |
| "compute incremental stats {0}.{1}".format(db_name, tbl_name), |
| "truncate table {0}.{1}".format(db_name, tbl_name)], |
| False: [ |
| "create table {0}.{1} like functional.alltypessmall " |
| "stored as parquet".format(db_name, tbl_name), |
| "create view {0}.{1} as select * from functional.alltypessmall " |
| "where year=2009".format(db_name, view_name), |
| # in case of rename we process it as drop+create and hence |
| # the events-skipped counter is not updated. Instead if this event is processed, |
| # it will increment the tables-added and tables-removed counters. |
| "alter table {0}.{1} rename to {0}.{2}".format(db_name, tbl_name, tbl2), |
| "alter table {0}.{1} rename to {0}.{2}".format(db_name, tbl2, tbl_name), |
| "alter view {0}.{1} rename to {0}.{2}".format(db_name, view_name, view2), |
| "alter view {0}.{1} rename to {0}.{2}".format(db_name, view2, view_name), |
| # ADD_PARTITION cases |
| # dynamic partition insert (creates new partitions) |
| "insert into table {0}.{1} partition (year,month) " |
| "select * from functional.alltypessmall where month % 2 = 0".format(db_name, |
| tbl_name), |
| "insert overwrite table {0}.{1} partition (year,month) " |
| "select * from functional.alltypessmall where month % 2 = 1".format(db_name, |
| tbl_name), |
| # we add this statement below just to make sure that the subsequent statement is |
| # a no-op |
| "alter table {0}.{1} add if not exists partition (year=2100, month=1)".format( |
| db_name, tbl_name), |
| "alter table {0}.{1} add if not exists partition (year=2100, month=1)".format( |
| db_name, tbl_name), |
| # DROP_PARTITION cases |
| "alter table {0}.{1} drop if exists partition (year=2100, month=1)".format( |
| db_name, tbl_name), |
| # drop non-existing partition; essentially this is a no-op |
| "alter table {0}.{1} drop if exists partition (year=2100, month=1)".format( |
| db_name, tbl_name), |
| # empty table case where no insert events are generated |
| "insert overwrite {0}.{1} select * from {0}.{1}".format( |
| db_name, empty_unpartitioned_tbl), |
| "insert overwrite {0}.{1} partition(part) select * from {0}.{1}".format( |
| db_name, empty_partitioned_tbl), |
| # in case of ACID tables no INSERT event is generated as the COMMIT event |
| # contains the related data |
| "create table {0}.{1} (c1 int) {2}".format(db_name, |
| acid_no_part_tbl_name, acid_props), |
| "insert into table {0}.{1} values (1) ".format(db_name, acid_no_part_tbl_name), |
| "insert overwrite table {0}.{1} select * from {0}.{1}".format( |
| db_name, acid_no_part_tbl_name), |
| "truncate table {0}.{1}".format(db_name, acid_no_part_tbl_name), |
| # the table is empty so the following insert adds 0 rows |
| "insert overwrite table {0}.{1} select * from {0}.{1}".format( |
| db_name, acid_no_part_tbl_name), |
| "create table {0}.{1} (c1 int) partitioned by (part int) {2}".format(db_name, |
| acid_tbl_name, acid_props), |
| "insert into table {0}.{1} partition (part=1) " |
| "values (1) ".format(db_name, acid_tbl_name), |
| "insert into table {0}.{1} partition (part) select id, int_col " |
| "from functional.alltypestiny".format(db_name, acid_tbl_name), |
| # repeat the same insert, now it writes to existing partitions |
| "insert into table {0}.{1} partition (part) select id, int_col " |
| "from functional.alltypestiny".format(db_name, acid_tbl_name), |
| # following insert overwrite is used instead of truncate, because truncate |
| # leads to a non-self event that reloads the table |
| "insert overwrite table {0}.{1} partition (part) select id, int_col " |
| "from functional.alltypestiny where id=-1".format(db_name, acid_tbl_name), |
| "insert overwrite table {0}.{1} partition (part) select id, int_col " |
| "from functional.alltypestiny".format(db_name, acid_tbl_name), |
| "insert overwrite {0}.{1} partition(part) select * from {0}.{1}".format( |
| db_name, acid_tbl_name), |
| # recover partitions will generate add_partition events |
| "alter table {0}.{1} recover partitions".format(db_name, recover_tbl_name) |
| ] |
| } |
| return self_event_test_queries |
| |
| def __get_hive_test_queries(self, db_name, recover_tbl_name): |
| tbl_name = ImpalaTestSuite.get_random_name("hive_test_tbl_") |
| tbl2 = ImpalaTestSuite.get_random_name("hive_renamed_tbl_") |
| view_name = ImpalaTestSuite.get_random_name("hive_view_") |
| # we use a custom table schema to make it easier to change columns later in the |
| # test_queries |
| self.client.execute("create table {0}.{1} (key int) partitioned by " |
| "(part int) stored as parquet".format(db_name, tbl_name)) |
| self.client.execute( |
| "create view {0}.{1} as select * from functional.alltypessmall where year=2009" |
| .format(db_name, view_name)) |
| # events-processor only refreshes loaded tables, hence its important to issue a |
| # refresh here so that table is in loaded state |
| self.client.execute("refresh {0}.{1}".format(db_name, tbl_name)) |
| self_event_test_queries = [ |
| # ALTER_DATABASE cases |
| "alter database {0} set dbproperties ('comment'='self-event test " |
| "database')".format(db_name), |
| "alter database {0} set owner user `test-user`".format(db_name), |
| # ALTER_TABLE case |
| "alter table {0}.{1} set tblproperties ('k'='v')".format(db_name, tbl_name), |
| "alter table {0}.{1} add columns (value string)".format(db_name, tbl_name), |
| "alter table {0}.{1} set owner user `test-user`".format(db_name, tbl_name), |
| "alter table {0}.{1} set owner role `test-role`".format(db_name, tbl_name), |
| # need to set this config to make sure the dynamic partition insert works below |
| "set hive.exec.dynamic.partition.mode=nonstrict", |
| # ADD_PARTITION cases |
| "insert into table {0}.{1} partition (part=2009) " |
| "select id as key, string_col as value from functional.alltypessmall".format( |
| db_name, tbl_name), |
| # add partition |
| "alter table {0}.{1} add if not exists partition (part=1111)".format( |
| db_name, tbl_name), |
| # add existing partition; essentially this is a no-op |
| "alter table {0}.{1} add if not exists partition (part=1111)".format( |
| db_name, tbl_name), |
| # DROP_PARTITION cases |
| "alter table {0}.{1} drop if exists partition (part=1111)".format( |
| db_name, tbl_name), |
| # drop non-existing partition; essentially this is a no-op |
| "alter table {0}.{1} drop if exists partition (part=1111)".format( |
| db_name, tbl_name), |
| # compute stats will generates ALTER_PARTITION |
| "analyze table {0}.{1} compute statistics for columns".format(db_name, tbl_name), |
| "msck repair table {0}.{1}".format(db_name, recover_tbl_name), |
| # we rename in the end since impala will have the new table in unloaded |
| # state after rename and hence any events later will be ignored anyways. |
| "alter table {0}.{1} rename to {0}.{2}".format(db_name, tbl_name, tbl2), |
| "alter view {0}.{1} rename to {0}.{2}".format(db_name, view_name, |
| ImpalaTestSuite.get_random_name("view_")), |
| ] |
| return self_event_test_queries |
| |
| @staticmethod |
| def __get_self_event_metrics(): |
| """ |
| Gets the tables-refreshed, partitions-refreshed and events-skipped metric values |
| from Metastore EventsProcessor |
| """ |
| tbls_refreshed_count = EventProcessorUtils.get_int_metric('tables-refreshed', 0) |
| partitions_refreshed_count = EventProcessorUtils.get_int_metric( |
| 'partitions-refreshed', 0) |
| events_skipped_count = EventProcessorUtils.get_int_metric('events-skipped', 0) |
| return int(tbls_refreshed_count), int(partitions_refreshed_count), \ |
| int(events_skipped_count) |
| |
| def __exec_sql_and_check_selfevent_counter(self, stmt, use_impala_client, |
| check_events_skipped_counter=True): |
| """ |
| Method runs a given query statement using a impala client or hive client based on the |
| argument use_impala_client and confirms if the self-event related counters are as |
| expected based on whether we expect a self-event or not. If the |
| check_self_event_counter is False it skips checking the events-skipped metric. |
| """ |
| EventProcessorUtils.wait_for_event_processing(self) |
| tbls_refreshed, partitions_refreshed, \ |
| events_skipped = self.__get_self_event_metrics() |
| last_synced_event = EventProcessorUtils.get_last_synced_event_id() |
| logging.info("Running statement in {1}: {0}".format(stmt, |
| "impala" if use_impala_client else "hive")) |
| if not use_impala_client: |
| self.run_stmt_in_hive(stmt) |
| else: |
| self.client.execute(stmt) |
| |
| EventProcessorUtils.wait_for_event_processing(self) |
| tbls_refreshed_after, partitions_refreshed_after, \ |
| events_skipped_after = self.__get_self_event_metrics() |
| last_synced_event_after = EventProcessorUtils.get_last_synced_event_id() |
| # we assume that any event which comes due to stmts run from impala-client are |
| # self-events |
| logging.info( |
| "Event id before {0} event id after {1}".format(last_synced_event, |
| last_synced_event_after)) |
| if use_impala_client: |
| # self-event counter must increase if this is a self-event if |
| # check_self_event_counter is set |
| # some of the test queries generate no events at all. If that is the case |
| # skip the below comparison |
| if last_synced_event_after > last_synced_event: |
| if check_events_skipped_counter: |
| assert events_skipped_after > events_skipped, \ |
| "Failing query(impala={}): {}".format(use_impala_client, stmt) |
| # if this is a self-event, no table or partitions should be refreshed |
| assert tbls_refreshed == tbls_refreshed_after, \ |
| "Failing query(impala={}): {}".format(use_impala_client, stmt) |
| assert partitions_refreshed == partitions_refreshed_after, \ |
| "Failing query(impala={}): {}".format(use_impala_client, stmt) |
| else: |
| # hive was used to run the stmts, any events generated should not have been deemed |
| # as self events unless there are empty partition add/drop events |
| assert events_skipped <= events_skipped_after |
| |
| def __get_tbl_location(self, db_name, tbl_name): |
| assert self.hive_client is not None |
| return self.hive_client.get_table(db_name, tbl_name).sd.location |
| |
| def __get_transactional_tblproperties(self, is_transactional): |
| """ |
| Util method to generate the tblproperties for transactional tables |
| """ |
| tblproperties = "" |
| if is_transactional: |
| tblproperties = "tblproperties ('transactional'='true'," \ |
| "'transactional_properties'='insert_only')" |
| return tblproperties |
| |
| @CustomClusterTestSuite.with_args(catalogd_args="--hms_event_polling_interval_s=5") |
| def test_stale_drop_partition_events(self, unique_database): |
| """Regression Tests for IMPALA-12256. Verifies stale DROP_PARTITION events are |
| skipped even if they are processed late after some other DDLs. Uses a higher polling |
| interval to ensure late processing on the events""" |
| self.client.execute( |
| "create table %s.part(i int) partitioned by (p int) stored as textfile" |
| % unique_database) |
| self.client.execute( |
| "insert into %s.part partition (p=0) values (0)" % unique_database) |
| # These DDLs will reload the partition metadata. We will verify they don't lose |
| # the create event ids after the reload. |
| partition_ddls = [ |
| "compute stats %s.part" % unique_database, |
| "compute incremental stats %s.part" % unique_database, |
| "compute incremental stats %s.part partition(p=0)" % unique_database, |
| "alter table %s.part partition(p=0) set row format" |
| " delimited fields terminated by ','" % unique_database, |
| "alter table %s.part partition(p=0) set fileformat parquet" % unique_database, |
| "alter table %s.part partition(p=0) set location '/tmp'" % unique_database, |
| "alter table %s.part partition(p=0) set tblproperties('k'='v')" % unique_database, |
| "refresh %s.part partition(p=0)" % unique_database, |
| "refresh %s.part" % unique_database, |
| ] |
| # Wait until the events in preparing the table are consumed. |
| EventProcessorUtils.wait_for_event_processing(self) |
| parts_added_before = EventProcessorUtils.get_int_metric("partitions-added") |
| parts_refreshed_before = EventProcessorUtils.get_int_metric("partitions-refreshed") |
| parts_removed_before = EventProcessorUtils.get_int_metric("partitions-removed") |
| for ddl in partition_ddls: |
| events_skipped_before = EventProcessorUtils.get_int_metric("events-skipped") |
| # Drop-create the partition and then runs a DDL on it. A DROP_PARTITION and an |
| # ADD_PARTITION event will be generated and should be skipped. The 3rd DDL might |
| # generate an ALTER_PARTITION event but it should be skipped as self-event. |
| # Note that we don't perform self-event detection on ADD/DROP_PARTITION events. |
| # They are skipped based on the partition level create event ids. So we should see |
| # no partitions are added/removed/refreshed if we correctly track the create event |
| # id (saved by the 2nd DDL that creates the partition). |
| # For the DROP_PARTITION event, there are 3 cases: |
| # 1) The DROP_PARTITION event is processed before the INSERT statement. |
| # It's skipped since the partition doesn't exist. |
| # 2) The DROP_PARTITION event is processed after the INSERT statement |
| # and before the 3rd DDL. The INSERT statement creates the partition so saves |
| # the create event id which is higher than the id of the DROP_PARTITION event. |
| # Thus the DROP_PARTITION event is skipped. |
| # 3) The DROP_PARTITION event is processed after the 3rd DDL. The reload triggered |
| # by the DDL should keep track of the create event id so the DROP_PARTITION event |
| # can be skipped. |
| # This test sets hms_event_polling_interval_s to 5 which is long enough for the |
| # 3 DDLs to finish. So it's more likely the 3rd case would happen, which is the |
| # case of IMPALA-12256. |
| self.client.execute( |
| "alter table %s.part drop partition (p=0)" % unique_database) |
| self.client.execute( |
| "insert into %s.part partition(p=0) values (1),(2)" % unique_database) |
| self.client.execute(ddl) |
| EventProcessorUtils.wait_for_event_processing(self) |
| events_skipped_after = EventProcessorUtils.get_int_metric("events-skipped") |
| parts_added_after = EventProcessorUtils.get_int_metric("partitions-added") |
| parts_refreshed_after = EventProcessorUtils.get_int_metric("partitions-refreshed") |
| parts_removed_after = EventProcessorUtils.get_int_metric("partitions-removed") |
| # Event-processor should not update any partitions since all events should be |
| # skipped |
| assert parts_removed_before == parts_removed_after |
| assert parts_added_before == parts_added_after |
| assert parts_refreshed_before == parts_refreshed_after |
| assert events_skipped_after > events_skipped_before |
| |
| @CustomClusterTestSuite.with_args(catalogd_args="--hms_event_polling_interval_s=5") |
| def test_truncate_table_from_hive(self, unique_database): |
| """IMPALA-12636: verify truncate table from hive reloads file metadata in Impala""" |
| hive_tbl = "tbl_in_hive" |
| values = "values (10),(20),(30)" |
| |
| def verify_truncate_op_in_hive(tbl_name, is_transactional, is_partitioned, |
| is_batched): |
| create_query = " ".join(["create", "table `{}`.`{}` (i int)", |
| " partitioned by (year int) " if is_partitioned else '', |
| self.__get_transactional_tblproperties(is_transactional)]) |
| self.execute_query(create_query.format(unique_database, tbl_name)) |
| insert_query = " ".join(["insert into `{}`.`{}`", "partition (year=2024)" |
| if is_partitioned else '', values]) |
| self.run_stmt_in_hive(insert_query.format(unique_database, tbl_name)) |
| EventProcessorUtils.wait_for_event_processing(self) |
| self.client.execute("refresh {}.{}".format(unique_database, tbl_name)) |
| truncate_query = " ".join(["truncate table `{}`.`{}`", "partition (year=2024)" |
| if is_partitioned else '']) |
| self.run_stmt_in_hive(truncate_query.format(unique_database, tbl_name)) |
| if is_batched: |
| self.run_stmt_in_hive( |
| "insert into {}.{} partition (year=2024) values (1),(2)" |
| .format(unique_database, tbl_name)) |
| EventProcessorUtils.wait_for_event_processing(self) |
| data = int(self.execute_scalar("select count(*) from {0}.{1}".format( |
| unique_database, tbl_name))) |
| assert data == 2 if is_batched else data == 0 |
| self.client.execute("drop table {}.{}".format(unique_database, tbl_name)) |
| # Case-I: truncate single partition |
| verify_truncate_op_in_hive(hive_tbl, False, False, False) |
| verify_truncate_op_in_hive(hive_tbl, True, False, False) |
| verify_truncate_op_in_hive(hive_tbl, False, True, False) |
| verify_truncate_op_in_hive(hive_tbl, False, True, True) |
| verify_truncate_op_in_hive(hive_tbl, True, True, False) |
| verify_truncate_op_in_hive(hive_tbl, True, True, True) |
| |
| # Case-II: truncate partition in multi partition |
| hive_tbl = "multi_part_tbl" |
| self.client.execute("create table {}.{} (i int) partitioned by " |
| "(p int, q int)".format(unique_database, hive_tbl)) |
| self.client.execute("insert into {}.{} partition(p, q) values " |
| "(0,0,0), (0,0,1), (0,0,2)".format(unique_database, hive_tbl)) |
| self.client.execute("insert into {}.{} partition(p, q) values " |
| "(0,1,0), (0,1,1)".format(unique_database, hive_tbl)) |
| self.run_stmt_in_hive("truncate table {}.{} partition(p=0)" |
| .format(unique_database, hive_tbl)) |
| EventProcessorUtils.wait_for_event_processing(self) |
| data = int(self.execute_scalar("select count(*) from {0}.{1}".format( |
| unique_database, hive_tbl))) |
| assert data == 2 |
| self.run_stmt_in_hive("truncate table {}.{}" |
| .format(unique_database, hive_tbl)) |
| EventProcessorUtils.wait_for_event_processing(self) |
| data = int(self.execute_scalar("select count(*) from {0}.{1}".format( |
| unique_database, hive_tbl))) |
| assert data == 0 |
| |
| @SkipIf.is_test_jdk |
| @CustomClusterTestSuite.with_args( |
| catalogd_args="--hms_event_polling_interval_s=100", |
| hive_conf_dir=HIVE_SITE_HOUSEKEEPING_ON) |
| def test_commit_compaction_with_abort_txn(self, unique_database): |
| """Use a long enough polling interval to allow Hive statements to finish before |
| the ABORT_TXN event is processed. In local tests, the Hive statements usually |
| finish in 60s. |
| TODO: improve this by adding commands to pause and resume the event-processor.""" |
| tbl = "part_table" |
| fq_tbl = unique_database + '.' + tbl |
| acid = AcidTxn(self.hive_client) |
| self.run_stmt_in_hive( |
| "create transactional table {} (i int) partitioned by (p int)".format(fq_tbl)) |
| |
| # Allocate a write id on this table and abort the txn |
| txn_id = acid.open_txns() |
| acid.allocate_table_write_ids(txn_id, unique_database, tbl) |
| acid.abort_txn(txn_id) |
| |
| # Insert some rows and trigger compaction |
| for i in range(2): |
| self.run_stmt_in_hive( |
| "insert into {} partition(p=0) values (1),(2),(3)".format(fq_tbl)) |
| self.run_stmt_in_hive( |
| "alter table {} partition(p=0) compact 'major' and wait".format(fq_tbl)) |
| |
| # The CREATE_TABLE event hasn't been processed yet so we have to explictily invalidate |
| # the table first. |
| self.client.execute("invalidate metadata " + fq_tbl) |
| # Reload the table so the latest valid writeIdList is loaded |
| self.client.execute("refresh " + fq_tbl) |
| # Process the ABORT_TXN event |
| EventProcessorUtils.wait_for_event_processing(self, timeout=100) |
| assert EventProcessorUtils.get_event_processor_status() == "ACTIVE" |
| # Uncomment this once we can stop and resume the event-processor using commands. |
| # Currently the test is flaky with it since the Hive statements could take longer to |
| # finish than 100s (e.g. I saw a run of 5mins). |
| # self.assert_catalogd_log_contains("INFO", "Not added ABORTED write id 1 since it's " |
| # + "not opened and might already be cleaned up") |
| |
| @CustomClusterTestSuite.with_args( |
| catalogd_args="--hms_event_incremental_refresh_transactional_table=false") |
| def test_no_hms_event_incremental_refresh_transactional_table(self, unique_database): |
| """IMPALA-12835: Test that Impala notices inserts to acid tables when |
| hms_event_incremental_refresh_transactional_table is false. |
| """ |
| for partitioned in [False, True]: |
| tbl = "part_tbl" if partitioned else "tbl" |
| fq_tbl = unique_database + '.' + tbl |
| part_create = " partitioned by (p int)" if partitioned else "" |
| part_insert = " partition (p = 1)" if partitioned else "" |
| |
| self.run_stmt_in_hive( |
| "create transactional table {} (i int){}".format(fq_tbl, part_create)) |
| EventProcessorUtils.wait_for_event_processing(self) |
| |
| # Load the table in Impala before INSERT |
| self.client.execute("refresh " + fq_tbl) |
| self.run_stmt_in_hive( |
| "insert into {}{} values (1),(2),(3)".format(fq_tbl, part_insert)) |
| EventProcessorUtils.wait_for_event_processing(self) |
| |
| results = self.client.execute("select i from " + fq_tbl) |
| assert results.data == ["1", "2", "3"] |
