tests/custom_cluster/test_event_processing.py - impala - Git at Google

 # Licensed to the Apache Software Foundation (ASF) under one
 # or more contributor license agreements.  See the NOTICE file
 # distributed with this work for additional information
 # regarding copyright ownership.  The ASF licenses this file
 # to you under the Apache License, Version 2.0 (the
 # "License"); you may not use this file except in compliance
 # with the License.  You may obtain a copy of the License at
 #
 #   http://www.apache.org/licenses/LICENSE-2.0
 #
 # Unless required by applicable law or agreed to in writing,
 # software distributed under the License is distributed on an
 # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.

 import pytest
 import json
 import time
 import requests

 from tests.common.environ import build_flavor_timeout
 from tests.common.skip import SkipIfS3, SkipIfABFS, SkipIfADLS, SkipIfIsilon, \
     SkipIfLocal, SkipIfHive2
 from tests.common.custom_cluster_test_suite import CustomClusterTestSuite
 from tests.common.skip import SkipIfS3, SkipIfABFS, SkipIfADLS, SkipIfIsilon, SkipIfLocal
 from tests.util.hive_utils import HiveDbWrapper
 from tests.util.event_processor_utils import EventProcessorUtils


 @SkipIfS3.hive
 @SkipIfABFS.hive
 @SkipIfADLS.hive
 @SkipIfIsilon.hive
 @SkipIfLocal.hive
 class TestEventProcessing(CustomClusterTestSuite):
   """This class contains tests that exercise the event processing mechanism in the
   catalog."""
   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()
       self.run_stmt_in_hive("create database testBlackListedDb")
       self.run_stmt_in_hive("create table testBlackListedDb.testtbl (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.hive_client)
       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")

   @pytest.mark.execute_serially
   @CustomClusterTestSuite.with_args(catalogd_args="--hms_event_polling_interval_s=2")
   @SkipIfHive2.acid
   def test_transactional_insert_events(self):
     """Executes 'run_test_insert_events' for transactional tables.
     """
     self.run_test_insert_events(is_transactional=True)

   @pytest.mark.execute_serially
   @CustomClusterTestSuite.with_args(catalogd_args="--hms_event_polling_interval_s=2")
   def test_insert_events(self):
     """Executes 'run_test_insert_events' for non-transactional tables.
     """
     self.run_test_insert_events()

   def run_test_insert_events(self, is_transactional=False):
     """Test for insert event processing. Events are created in Hive and processed in
     Impala. The following cases are tested :
     Insert into table --> for partitioned and non-partitioned table
     Insert overwrite table --> for partitioned and non-partitioned table
     Insert into partition --> for partitioned table
     """
     db_name = 'test_db'
     with HiveDbWrapper(self, db_name):
      # Test table with no partitions.
      TBL_INSERT_NOPART = 'tbl_insert_nopart'
      self.run_stmt_in_hive("drop table if exists %s.%s" % (db_name, TBL_INSERT_NOPART))
      last_synced_event_id = self.get_last_synced_event_id()
      TBLPROPERTIES = ""
      if is_transactional:
        TBLPROPERTIES = "TBLPROPERTIES ('transactional'='true'," \
            "'transactional_properties'='insert_only')"
      self.run_stmt_in_hive("create table %s.%s (id int, val int) %s"
          % (db_name, TBL_INSERT_NOPART, TBLPROPERTIES))
      # Test insert into table, this will fire an insert event.
      self.run_stmt_in_hive("insert into %s.%s values(101, 200)"
          % (db_name, TBL_INSERT_NOPART))
      # With MetastoreEventProcessor running, the insert event will be processed. Query the
      # table from Impala.
      assert self.wait_for_insert_event_processing(last_synced_event_id) is True
      # Verify that the data is present in Impala.
      data = self.execute_scalar("select * from %s.%s" % (db_name, TBL_INSERT_NOPART))
      assert data.split('\t') == ['101', '200']

      # Test insert overwrite. Overwrite the existing value.
      last_synced_event_id = self.get_last_synced_event_id()
      self.run_stmt_in_hive("insert overwrite table %s.%s values(101, 201)"
          % (db_name, TBL_INSERT_NOPART))
      # Make sure the event has been processed.
      assert self.wait_for_insert_event_processing(last_synced_event_id) is True
      # Verify that the data is present in Impala.
      data = self.execute_scalar("select * from %s.%s" % (db_name, TBL_INSERT_NOPART))
      assert data.split('\t') == ['101', '201']

      # Test partitioned table.
      last_synced_event_id = self.get_last_synced_event_id()
      TBL_INSERT_PART = 'tbl_insert_part'
      self.run_stmt_in_hive("drop table if exists %s.%s" % (db_name, TBL_INSERT_PART))
      self.run_stmt_in_hive("create table %s.%s (id int, name string) "
          "partitioned by(day int, month int, year int) %s"
          % (db_name, TBL_INSERT_PART, TBLPROPERTIES))
      # Insert data into partitions.
      self.run_stmt_in_hive("insert into %s.%s partition(day=28, month=03, year=2019)"
          "values(101, 'x')" % (db_name, TBL_INSERT_PART))
      # Make sure the event has been processed.
      assert self.wait_for_insert_event_processing(last_synced_event_id) is True
      # Verify that the data is present in Impala.
      data = self.execute_scalar("select * from %s.%s" % (db_name, TBL_INSERT_PART))
      assert data.split('\t') == ['101', 'x', '28', '3', '2019']

      # Test inserting into existing partitions.
      last_synced_event_id = self.get_last_synced_event_id()
      self.run_stmt_in_hive("insert into %s.%s partition(day=28, month=03, year=2019)"
          "values(102, 'y')" % (db_name, TBL_INSERT_PART))
      assert self.wait_for_insert_event_processing(last_synced_event_id) is True
      # Verify that the data is present in Impala.
      data = self.execute_scalar("select count(*) from %s.%s where day=28 and month=3 "
          "and year=2019" % (db_name, TBL_INSERT_PART))
      assert data.split('\t') == ['2']

      # Test insert overwrite into existing partitions
      last_synced_event_id = self.get_last_synced_event_id()
      self.run_stmt_in_hive("insert overwrite table %s.%s partition(day=28, month=03, "
          "year=2019)" "values(101, 'z')" % (db_name, TBL_INSERT_PART))
      assert self.wait_for_insert_event_processing(last_synced_event_id) is True
      # Verify that the data is present in Impala.
      data = self.execute_scalar("select * from %s.%s where day=28 and month=3 and"
          " year=2019 and id=101" % (db_name, TBL_INSERT_PART))
      assert data.split('\t') == ['101', 'z', '28', '3', '2019']

   @CustomClusterTestSuite.with_args(
     catalogd_args="--hms_event_polling_interval_s=1"
   )
   @SkipIfHive2.acid
   def test_empty_partition_events_transactional(self, unique_database):
     self._run_test_empty_partition_events(unique_database, True)

   @CustomClusterTestSuite.with_args(
     catalogd_args="--hms_event_polling_interval_s=1"
   )
   def test_empty_partition_events(self, unique_database):
     self._run_test_empty_partition_events(unique_database, False)

   def _run_test_empty_partition_events(self, unique_database, is_transactional):
     TBLPROPERTIES = ""
     if is_transactional:
        TBLPROPERTIES = "TBLPROPERTIES ('transactional'='true'," \
            "'transactional_properties'='insert_only')"
     test_tbl = unique_database + ".test_events"
     self.run_stmt_in_hive("create table {0} (key string, value string) \
       partitioned by (year int) stored as parquet {1}".format(test_tbl, TBLPROPERTIES))
     EventProcessorUtils.wait_for_event_processing(self.hive_client)
     self.client.execute("describe {0}".format(test_tbl))

     self.run_stmt_in_hive(
       "alter table {0} add partition (year=2019)".format(test_tbl))
     EventProcessorUtils.wait_for_event_processing(self.hive_client)
     assert [('2019',)] == self.get_impala_partition_info(test_tbl, 'year')

     self.run_stmt_in_hive(
       "alter table {0} add if not exists partition (year=2019)".format(test_tbl))
     EventProcessorUtils.wait_for_event_processing(self.hive_client)
     assert [('2019',)] == self.get_impala_partition_info(test_tbl, 'year')
     assert EventProcessorUtils.get_event_processor_status() == "ACTIVE"

     self.run_stmt_in_hive(
       "alter table {0} drop partition (year=2019)".format(test_tbl))
     EventProcessorUtils.wait_for_event_processing(self.hive_client)
     assert ('2019') not in self.get_impala_partition_info(test_tbl, 'year')
     assert EventProcessorUtils.get_event_processor_status() == "ACTIVE"

     self.run_stmt_in_hive(
       "alter table {0} drop if exists partition (year=2019)".format(test_tbl))
     EventProcessorUtils.wait_for_event_processing(self.hive_client)
     assert ('2019') not in self.get_impala_partition_info(test_tbl, 'year')
     assert EventProcessorUtils.get_event_processor_status() == "ACTIVE"


   def wait_for_insert_event_processing(self, previous_event_id):
     """Waits until the event processor has finished processing insert events. Since two
     events are created for every insert done through hive, we wait until the event id is
     incremented by at least two. Returns true if at least two events were processed within
     self.PROCESSING_TIMEOUT_S, False otherwise.
     """
     new_event_id = self.get_last_synced_event_id()
     success = False
     end_time = time.time() + self.PROCESSING_TIMEOUT_S
     while time.time() < end_time:
       new_event_id = self.get_last_synced_event_id()
       if new_event_id - previous_event_id >= 2:
         success = True
         break
       time.sleep(0.1)
     # Wait for catalog update to be propagated.
     time.sleep(build_flavor_timeout(2, slow_build_timeout=4))
     return success

   def get_event_processor_metrics(self):
     """Scrapes the catalog's /events webpage and return a dictionary with the event
     processor metrics."""
     response = requests.get("%s/events?json" % self.CATALOG_URL)
     assert response.status_code == requests.codes.ok
     varz_json = json.loads(response.text)
     metrics = varz_json["event_processor_metrics"].strip().splitlines()

     # Helper to strip a pair of elements
     def strip_pair(p):
       return (p[0].strip(), p[1].strip())

     pairs = [strip_pair(kv.split(':')) for kv in metrics if kv]
     return dict(pairs)

   def get_last_synced_event_id(self):
     """Returns the last_synced_event_id."""
     metrics = self.get_event_processor_metrics()
     assert 'last-synced-event-id' in metrics.keys()
     return int(metrics['last-synced-event-id'])
	# Licensed to the Apache Software Foundation (ASF) under one
	# or more contributor license agreements. See the NOTICE file
	# distributed with this work for additional information
	# regarding copyright ownership. The ASF licenses this file
	# to you under the Apache License, Version 2.0 (the
	# "License"); you may not use this file except in compliance
	# with the License. You may obtain a copy of the License at
	#
	# http://www.apache.org/licenses/LICENSE-2.0
	#
	# Unless required by applicable law or agreed to in writing,
	# software distributed under the License is distributed on an
	# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	# KIND, either express or implied. See the License for the
	# specific language governing permissions and limitations
	# under the License.

	import pytest
	import json
	import time
	import requests

	from tests.common.environ import build_flavor_timeout
	from tests.common.skip import SkipIfS3, SkipIfABFS, SkipIfADLS, SkipIfIsilon, \
	SkipIfLocal, SkipIfHive2
	from tests.common.custom_cluster_test_suite import CustomClusterTestSuite
	from tests.common.skip import SkipIfS3, SkipIfABFS, SkipIfADLS, SkipIfIsilon, SkipIfLocal
	from tests.util.hive_utils import HiveDbWrapper
	from tests.util.event_processor_utils import EventProcessorUtils


	@SkipIfS3.hive
	@SkipIfABFS.hive
	@SkipIfADLS.hive
	@SkipIfIsilon.hive
	@SkipIfLocal.hive
	class TestEventProcessing(CustomClusterTestSuite):
	"""This class contains tests that exercise the event processing mechanism in the
	catalog."""
	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()
	self.run_stmt_in_hive("create database testBlackListedDb")
	self.run_stmt_in_hive("create table testBlackListedDb.testtbl (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.hive_client)
	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")

	@pytest.mark.execute_serially
	@CustomClusterTestSuite.with_args(catalogd_args="--hms_event_polling_interval_s=2")
	@SkipIfHive2.acid
	def test_transactional_insert_events(self):
	"""Executes 'run_test_insert_events' for transactional tables.
	"""
	self.run_test_insert_events(is_transactional=True)

	@pytest.mark.execute_serially
	@CustomClusterTestSuite.with_args(catalogd_args="--hms_event_polling_interval_s=2")
	def test_insert_events(self):
	"""Executes 'run_test_insert_events' for non-transactional tables.
	"""
	self.run_test_insert_events()

	def run_test_insert_events(self, is_transactional=False):
	"""Test for insert event processing. Events are created in Hive and processed in
	Impala. The following cases are tested :
	Insert into table --> for partitioned and non-partitioned table
	Insert overwrite table --> for partitioned and non-partitioned table
	Insert into partition --> for partitioned table
	"""
	db_name = 'test_db'
	with HiveDbWrapper(self, db_name):
	# Test table with no partitions.
	TBL_INSERT_NOPART = 'tbl_insert_nopart'
	self.run_stmt_in_hive("drop table if exists %s.%s" % (db_name, TBL_INSERT_NOPART))
	last_synced_event_id = self.get_last_synced_event_id()
	TBLPROPERTIES = ""
	if is_transactional:
	TBLPROPERTIES = "TBLPROPERTIES ('transactional'='true'," \
	"'transactional_properties'='insert_only')"
	self.run_stmt_in_hive("create table %s.%s (id int, val int) %s"
	% (db_name, TBL_INSERT_NOPART, TBLPROPERTIES))
	# Test insert into table, this will fire an insert event.
	self.run_stmt_in_hive("insert into %s.%s values(101, 200)"
	% (db_name, TBL_INSERT_NOPART))
	# With MetastoreEventProcessor running, the insert event will be processed. Query the
	# table from Impala.
	assert self.wait_for_insert_event_processing(last_synced_event_id) is True
	# Verify that the data is present in Impala.
	data = self.execute_scalar("select * from %s.%s" % (db_name, TBL_INSERT_NOPART))
	assert data.split('\t') == ['101', '200']

	# Test insert overwrite. Overwrite the existing value.
	last_synced_event_id = self.get_last_synced_event_id()
	self.run_stmt_in_hive("insert overwrite table %s.%s values(101, 201)"
	% (db_name, TBL_INSERT_NOPART))
	# Make sure the event has been processed.
	assert self.wait_for_insert_event_processing(last_synced_event_id) is True
	# Verify that the data is present in Impala.
	data = self.execute_scalar("select * from %s.%s" % (db_name, TBL_INSERT_NOPART))
	assert data.split('\t') == ['101', '201']

	# Test partitioned table.
	last_synced_event_id = self.get_last_synced_event_id()
	TBL_INSERT_PART = 'tbl_insert_part'
	self.run_stmt_in_hive("drop table if exists %s.%s" % (db_name, TBL_INSERT_PART))
	self.run_stmt_in_hive("create table %s.%s (id int, name string) "
	"partitioned by(day int, month int, year int) %s"
	% (db_name, TBL_INSERT_PART, TBLPROPERTIES))
	# Insert data into partitions.
	self.run_stmt_in_hive("insert into %s.%s partition(day=28, month=03, year=2019)"
	"values(101, 'x')" % (db_name, TBL_INSERT_PART))
	# Make sure the event has been processed.
	assert self.wait_for_insert_event_processing(last_synced_event_id) is True
	# Verify that the data is present in Impala.
	data = self.execute_scalar("select * from %s.%s" % (db_name, TBL_INSERT_PART))
	assert data.split('\t') == ['101', 'x', '28', '3', '2019']

	# Test inserting into existing partitions.
	last_synced_event_id = self.get_last_synced_event_id()
	self.run_stmt_in_hive("insert into %s.%s partition(day=28, month=03, year=2019)"
	"values(102, 'y')" % (db_name, TBL_INSERT_PART))
	assert self.wait_for_insert_event_processing(last_synced_event_id) is True
	# Verify that the data is present in Impala.
	data = self.execute_scalar("select count(*) from %s.%s where day=28 and month=3 "
	"and year=2019" % (db_name, TBL_INSERT_PART))
	assert data.split('\t') == ['2']

	# Test insert overwrite into existing partitions
	last_synced_event_id = self.get_last_synced_event_id()
	self.run_stmt_in_hive("insert overwrite table %s.%s partition(day=28, month=03, "
	"year=2019)" "values(101, 'z')" % (db_name, TBL_INSERT_PART))
	assert self.wait_for_insert_event_processing(last_synced_event_id) is True
	# Verify that the data is present in Impala.
	data = self.execute_scalar("select * from %s.%s where day=28 and month=3 and"
	" year=2019 and id=101" % (db_name, TBL_INSERT_PART))
	assert data.split('\t') == ['101', 'z', '28', '3', '2019']

	@CustomClusterTestSuite.with_args(
	catalogd_args="--hms_event_polling_interval_s=1"
	)
	@SkipIfHive2.acid
	def test_empty_partition_events_transactional(self, unique_database):
	self._run_test_empty_partition_events(unique_database, True)

	@CustomClusterTestSuite.with_args(
	catalogd_args="--hms_event_polling_interval_s=1"
	)
	def test_empty_partition_events(self, unique_database):
	self._run_test_empty_partition_events(unique_database, False)

	def _run_test_empty_partition_events(self, unique_database, is_transactional):
	TBLPROPERTIES = ""
	if is_transactional:
	TBLPROPERTIES = "TBLPROPERTIES ('transactional'='true'," \
	"'transactional_properties'='insert_only')"
	test_tbl = unique_database + ".test_events"
	self.run_stmt_in_hive("create table {0} (key string, value string) \
	partitioned by (year int) stored as parquet {1}".format(test_tbl, TBLPROPERTIES))
	EventProcessorUtils.wait_for_event_processing(self.hive_client)
	self.client.execute("describe {0}".format(test_tbl))

	self.run_stmt_in_hive(
	"alter table {0} add partition (year=2019)".format(test_tbl))
	EventProcessorUtils.wait_for_event_processing(self.hive_client)
	assert [('2019',)] == self.get_impala_partition_info(test_tbl, 'year')

	self.run_stmt_in_hive(
	"alter table {0} add if not exists partition (year=2019)".format(test_tbl))
	EventProcessorUtils.wait_for_event_processing(self.hive_client)
	assert [('2019',)] == self.get_impala_partition_info(test_tbl, 'year')
	assert EventProcessorUtils.get_event_processor_status() == "ACTIVE"

	self.run_stmt_in_hive(
	"alter table {0} drop partition (year=2019)".format(test_tbl))
	EventProcessorUtils.wait_for_event_processing(self.hive_client)
	assert ('2019') not in self.get_impala_partition_info(test_tbl, 'year')
	assert EventProcessorUtils.get_event_processor_status() == "ACTIVE"

	self.run_stmt_in_hive(
	"alter table {0} drop if exists partition (year=2019)".format(test_tbl))
	EventProcessorUtils.wait_for_event_processing(self.hive_client)
	assert ('2019') not in self.get_impala_partition_info(test_tbl, 'year')
	assert EventProcessorUtils.get_event_processor_status() == "ACTIVE"


	def wait_for_insert_event_processing(self, previous_event_id):
	"""Waits until the event processor has finished processing insert events. Since two
	events are created for every insert done through hive, we wait until the event id is
	incremented by at least two. Returns true if at least two events were processed within
	self.PROCESSING_TIMEOUT_S, False otherwise.
	"""
	new_event_id = self.get_last_synced_event_id()
	success = False
	end_time = time.time() + self.PROCESSING_TIMEOUT_S
	while time.time() < end_time:
	new_event_id = self.get_last_synced_event_id()
	if new_event_id - previous_event_id >= 2:
	success = True
	break
	time.sleep(0.1)
	# Wait for catalog update to be propagated.
	time.sleep(build_flavor_timeout(2, slow_build_timeout=4))
	return success

	def get_event_processor_metrics(self):
	"""Scrapes the catalog's /events webpage and return a dictionary with the event
	processor metrics."""
	response = requests.get("%s/events?json" % self.CATALOG_URL)
	assert response.status_code == requests.codes.ok
	varz_json = json.loads(response.text)
	metrics = varz_json["event_processor_metrics"].strip().splitlines()

	# Helper to strip a pair of elements
	def strip_pair(p):
	return (p[0].strip(), p[1].strip())

	pairs = [strip_pair(kv.split(':')) for kv in metrics if kv]
	return dict(pairs)

	def get_last_synced_event_id(self):
	"""Returns the last_synced_event_id."""
	metrics = self.get_event_processor_metrics()
	assert 'last-synced-event-id' in metrics.keys()
	return int(metrics['last-synced-event-id'])