tests/metadata/test_last_ddl_time_update.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.

 # Impala tests for DDL statements
 import time

 from tests.common.impala_test_suite import ImpalaTestSuite
 from tests.util.filesystem_utils import WAREHOUSE, IS_S3

 # Checks different statements' effect on last DDL time and last compute stats time.
 class TestLastDdlTimeUpdate(ImpalaTestSuite):

   @classmethod
   def get_workload(self):
     return 'functional-query'

   @classmethod
   def add_test_dimensions(cls):
     super(TestLastDdlTimeUpdate, cls).add_test_dimensions()
     # There is no reason to run these tests using all dimensions.
     cls.ImpalaTestMatrix.add_constraint(lambda v:\
         v.get_value('table_format').file_format == 'text' and\
         v.get_value('table_format').compression_codec == 'none')

     if cls.exploration_strategy() == 'core' and not IS_S3:
       # Don't run on core.  This test is very slow and we are unlikely
       # to regress here.
       cls.ImpalaTestMatrix.add_constraint(lambda v: False)

   # Convenience class to make calls to TestLastDdlTimeUpdate.run_test() shorter by
   # storing common arguments as members and substituting table name and HDFS warehouse
   # path to the query string.
   class TestHelper:
     def __init__(self, test_suite, db_name, tbl_name):
       self.test_suite = test_suite
       self.db_name = db_name
       self.tbl_name = tbl_name
       self.fq_tbl_name = "%s.%s" % (self.db_name, self.tbl_name)

     def expect_no_time_change(self, query):
       """Neither transient_lastDdlTime or impala.lastComputeStatsTime should be
       changed by running the query.
       The following strings are substituted in the query: "%(TBL)s" and "%(WAREHOUSE)s"
       """
       self.run_test(query, False, False)

     def expect_ddl_time_change(self, query):
       """Running the query should increase transient_lastDdlTime but
       not impala.lastComputeStatsTime.
       The following strings are substituted in the query: "%(TBL)s" and "%(WAREHOUSE)s"
       """
       self.run_test(query, True, False)

     def expect_stat_time_change(self, query):
       """Running the query should increase impala.lastComputeStatsTime but
       not transient_lastDdlTime.
       The following strings are substituted in the query: "%(TBL)s" and "%(WAREHOUSE)s"
       """
       self.run_test(query, False, True)

     def run_test(self, query, expect_changed_ddl_time, expect_changed_stats_time):
       """
       Runs the query and compares the last ddl/compute stats time before and after
       executing the query. If expect_changed_ddl_time or expect_changed_stat_time
       is true, then we expect the given table property to be increased, otherwise we
       expect it to be unchanged.
       The following strings are substituted in the query: "%(TBL)s" and "%(WAREHOUSE)s"
       """

       HIVE_LAST_DDL_TIME_PARAM_KEY = "transient_lastDdlTime"
       LAST_COMPUTE_STATS_TIME_KEY = "impala.lastComputeStatsTime"

       # Get timestamps before executing query.
       table = self.test_suite.hive_client.get_table(self.db_name, self.tbl_name)
       assert table is not None
       beforeDdlTime = table.parameters[HIVE_LAST_DDL_TIME_PARAM_KEY]
       beforeStatsTime = table.parameters[LAST_COMPUTE_STATS_TIME_KEY]
       # HMS uses a seconds granularity on the last ddl time - sleeping 1100 ms should be
       # enough to ensure that the new timestamps are strictly greater than the old ones.
       time.sleep (1.1)

       self.test_suite.execute_query(query %
         {'TBL': self.fq_tbl_name, 'WAREHOUSE': WAREHOUSE})

       # Get timestamps after executing query.
       table = self.test_suite.hive_client.get_table(self.db_name, self.tbl_name)
       afterDdlTime = table.parameters[HIVE_LAST_DDL_TIME_PARAM_KEY]
       afterStatsTime = table.parameters[LAST_COMPUTE_STATS_TIME_KEY]

       if expect_changed_ddl_time:
         # check that the new ddlTime is strictly greater than the old one.
         assert long(afterDdlTime) > long(beforeDdlTime)
       else:
         assert long(afterDdlTime) == long(beforeDdlTime)

       if expect_changed_stats_time:
         # check that the new statsTime is strictly greater than the old one.
         assert long(afterStatsTime) > long(beforeStatsTime)
       else:
         assert long(afterStatsTime) == long(beforeStatsTime)

   def test_alter(self, vector, unique_database):
     TBL_NAME = "alter_test_tbl"
     FQ_TBL_NAME = unique_database + "." + TBL_NAME

     self.execute_query("create external table %s (i int) "
                        "partitioned by (j int, s string)" % FQ_TBL_NAME)

     # compute statistics to fill table property impala.lastComputeStatsTime
     self.execute_query("compute stats %s" % FQ_TBL_NAME)

     h = TestLastDdlTimeUpdate.TestHelper(self, unique_database, TBL_NAME)

     # add/drop partitions
     h.expect_no_time_change("alter table %(TBL)s add partition (j=1, s='2012')")
     h.expect_no_time_change("alter table %(TBL)s add if not exists "
                             "partition (j=1, s='2012')")
     h.expect_no_time_change("alter table %(TBL)s drop partition (j=1, s='2012')")
     h.expect_no_time_change("alter table %(TBL)s drop if exists "
                             "partition (j=2, s='2012')")
     # change location of table
     h.expect_ddl_time_change("alter table %(TBL)s set location '%(WAREHOUSE)s'")
     # change format of table
     h.expect_ddl_time_change("alter table %(TBL)s set fileformat textfile")

     self.run_common_test_cases(h)

     # prepare for incremental statistics tests
     self.execute_query("drop stats %s" % FQ_TBL_NAME)
     self.execute_query("alter table %s add partition (j=1, s='2012')" % FQ_TBL_NAME)

     # compute incremental statistics
     h.expect_stat_time_change("compute incremental stats %(TBL)s")
     h.expect_stat_time_change("compute incremental stats %(TBL)s "
                               "partition (j=1, s='2012')")

     # drop incremental statistics
     h.expect_no_time_change("drop incremental stats %(TBL)s partition (j=1, s='2012')")

     # prepare for table sample statistics tests
     self.execute_query(
         "alter table %s set tblproperties ('impala.enable.stats.extrapolation'='true')"
         % FQ_TBL_NAME)

     # compute sampled statistics
     h.expect_stat_time_change("compute stats %(TBL)s tablesample system(20)")


   def test_insert(self, vector, unique_database):
     TBL_NAME = "insert_test_tbl"
     FQ_TBL_NAME = unique_database + "." + TBL_NAME
     self.execute_query("create external table %s (i int) "
                        "partitioned by (j int, s string)" % FQ_TBL_NAME)

     # initialize compute stats time
     self.execute_query("compute stats %s" % FQ_TBL_NAME)

     h = TestLastDdlTimeUpdate.TestHelper(self, unique_database, TBL_NAME)

     # static partition insert
     h.expect_no_time_change("insert into %(TBL)s partition(j=1, s='2012') select 10")
     # dynamic partition insert
     h.expect_no_time_change("insert into %(TBL)s partition(j, s) select 10, 2, '2013'")

     # dynamic partition insert changing no partitions (empty input)
     h.expect_no_time_change("insert into %(TBL)s partition(j, s) "
                             "select * from (select 10 as i, 2 as j, '2013' as s) as t "
                             "where t.i < 10")
     # dynamic partition insert modifying an existing partition
     h.expect_no_time_change("insert into %(TBL)s partition(j, s) select 20, 1, '2012'")

   def test_kudu(self, vector, unique_database):
     TBL_NAME = "kudu_test_tbl"
     FQ_TBL_NAME = unique_database + "." + TBL_NAME
     self.execute_query("create table %s (i int primary key) "
                        "partition by hash(i) partitions 3 stored as kudu" % FQ_TBL_NAME)

     # initialize last compute stats time
     self.execute_query("compute stats %s" % FQ_TBL_NAME)

     h = TestLastDdlTimeUpdate.TestHelper(self, unique_database, TBL_NAME)

     # insert
     h.expect_no_time_change("insert into %s values (1)" % FQ_TBL_NAME)

     self.run_common_test_cases(h)

   # Tests that should behave the same with HDFS and Kudu tables.
   def run_common_test_cases(self, test_helper):
     h = test_helper
     # rename columns
     h.expect_ddl_time_change("alter table %(TBL)s change column i k int")
     h.expect_ddl_time_change("alter table %(TBL)s change column k i int")
     # change table property
     h.expect_ddl_time_change("alter table %(TBL)s set tblproperties ('a'='b')")

     # changing table statistics manually
     h.expect_ddl_time_change("alter table %(TBL)s set tblproperties ('numRows'='1')")

     # changing column statistics manually
     h.expect_no_time_change("alter table %(TBL)s set column stats i ('numdvs'='3')")

     # compute statistics
     h.expect_stat_time_change("compute stats %(TBL)s")
     # compute statistics for a single column
     h.expect_stat_time_change("compute stats %(TBL)s (i)")

     # drop statistics
     h.expect_no_time_change("drop stats %(TBL)s")

     # invalidate metadata and reload table
     self.execute_query("invalidate metadata %s" % h.fq_tbl_name)
     # run any query to reload the table
     h.expect_no_time_change("describe %(TBL)s")
	# 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 DDL statements
	import time

	from tests.common.impala_test_suite import ImpalaTestSuite
	from tests.util.filesystem_utils import WAREHOUSE, IS_S3

	# Checks different statements' effect on last DDL time and last compute stats time.
	class TestLastDdlTimeUpdate(ImpalaTestSuite):

	@classmethod
	def get_workload(self):
	return 'functional-query'

	@classmethod
	def add_test_dimensions(cls):
	super(TestLastDdlTimeUpdate, cls).add_test_dimensions()
	# There is no reason to run these tests using all dimensions.
	cls.ImpalaTestMatrix.add_constraint(lambda v:\
	v.get_value('table_format').file_format == 'text' and\
	v.get_value('table_format').compression_codec == 'none')

	if cls.exploration_strategy() == 'core' and not IS_S3:
	# Don't run on core. This test is very slow and we are unlikely
	# to regress here.
	cls.ImpalaTestMatrix.add_constraint(lambda v: False)

	# Convenience class to make calls to TestLastDdlTimeUpdate.run_test() shorter by
	# storing common arguments as members and substituting table name and HDFS warehouse
	# path to the query string.
	class TestHelper:
	def __init__(self, test_suite, db_name, tbl_name):
	self.test_suite = test_suite
	self.db_name = db_name
	self.tbl_name = tbl_name
	self.fq_tbl_name = "%s.%s" % (self.db_name, self.tbl_name)

	def expect_no_time_change(self, query):
	"""Neither transient_lastDdlTime or impala.lastComputeStatsTime should be
	changed by running the query.
	The following strings are substituted in the query: "%(TBL)s" and "%(WAREHOUSE)s"
	"""
	self.run_test(query, False, False)

	def expect_ddl_time_change(self, query):
	"""Running the query should increase transient_lastDdlTime but
	not impala.lastComputeStatsTime.
	The following strings are substituted in the query: "%(TBL)s" and "%(WAREHOUSE)s"
	"""
	self.run_test(query, True, False)

	def expect_stat_time_change(self, query):
	"""Running the query should increase impala.lastComputeStatsTime but
	not transient_lastDdlTime.
	The following strings are substituted in the query: "%(TBL)s" and "%(WAREHOUSE)s"
	"""
	self.run_test(query, False, True)

	def run_test(self, query, expect_changed_ddl_time, expect_changed_stats_time):
	"""
	Runs the query and compares the last ddl/compute stats time before and after
	executing the query. If expect_changed_ddl_time or expect_changed_stat_time
	is true, then we expect the given table property to be increased, otherwise we
	expect it to be unchanged.
	The following strings are substituted in the query: "%(TBL)s" and "%(WAREHOUSE)s"
	"""

	HIVE_LAST_DDL_TIME_PARAM_KEY = "transient_lastDdlTime"
	LAST_COMPUTE_STATS_TIME_KEY = "impala.lastComputeStatsTime"

	# Get timestamps before executing query.
	table = self.test_suite.hive_client.get_table(self.db_name, self.tbl_name)
	assert table is not None
	beforeDdlTime = table.parameters[HIVE_LAST_DDL_TIME_PARAM_KEY]
	beforeStatsTime = table.parameters[LAST_COMPUTE_STATS_TIME_KEY]
	# HMS uses a seconds granularity on the last ddl time - sleeping 1100 ms should be
	# enough to ensure that the new timestamps are strictly greater than the old ones.
	time.sleep (1.1)

	self.test_suite.execute_query(query %
	{'TBL': self.fq_tbl_name, 'WAREHOUSE': WAREHOUSE})

	# Get timestamps after executing query.
	table = self.test_suite.hive_client.get_table(self.db_name, self.tbl_name)
	afterDdlTime = table.parameters[HIVE_LAST_DDL_TIME_PARAM_KEY]
	afterStatsTime = table.parameters[LAST_COMPUTE_STATS_TIME_KEY]

	if expect_changed_ddl_time:
	# check that the new ddlTime is strictly greater than the old one.
	assert long(afterDdlTime) > long(beforeDdlTime)
	else:
	assert long(afterDdlTime) == long(beforeDdlTime)

	if expect_changed_stats_time:
	# check that the new statsTime is strictly greater than the old one.
	assert long(afterStatsTime) > long(beforeStatsTime)
	else:
	assert long(afterStatsTime) == long(beforeStatsTime)

	def test_alter(self, vector, unique_database):
	TBL_NAME = "alter_test_tbl"
	FQ_TBL_NAME = unique_database + "." + TBL_NAME

	self.execute_query("create external table %s (i int) "
	"partitioned by (j int, s string)" % FQ_TBL_NAME)

	# compute statistics to fill table property impala.lastComputeStatsTime
	self.execute_query("compute stats %s" % FQ_TBL_NAME)

	h = TestLastDdlTimeUpdate.TestHelper(self, unique_database, TBL_NAME)

	# add/drop partitions
	h.expect_no_time_change("alter table %(TBL)s add partition (j=1, s='2012')")
	h.expect_no_time_change("alter table %(TBL)s add if not exists "
	"partition (j=1, s='2012')")
	h.expect_no_time_change("alter table %(TBL)s drop partition (j=1, s='2012')")
	h.expect_no_time_change("alter table %(TBL)s drop if exists "
	"partition (j=2, s='2012')")
	# change location of table
	h.expect_ddl_time_change("alter table %(TBL)s set location '%(WAREHOUSE)s'")
	# change format of table
	h.expect_ddl_time_change("alter table %(TBL)s set fileformat textfile")

	self.run_common_test_cases(h)

	# prepare for incremental statistics tests
	self.execute_query("drop stats %s" % FQ_TBL_NAME)
	self.execute_query("alter table %s add partition (j=1, s='2012')" % FQ_TBL_NAME)

	# compute incremental statistics
	h.expect_stat_time_change("compute incremental stats %(TBL)s")
	h.expect_stat_time_change("compute incremental stats %(TBL)s "
	"partition (j=1, s='2012')")

	# drop incremental statistics
	h.expect_no_time_change("drop incremental stats %(TBL)s partition (j=1, s='2012')")

	# prepare for table sample statistics tests
	self.execute_query(
	"alter table %s set tblproperties ('impala.enable.stats.extrapolation'='true')"
	% FQ_TBL_NAME)

	# compute sampled statistics
	h.expect_stat_time_change("compute stats %(TBL)s tablesample system(20)")


	def test_insert(self, vector, unique_database):
	TBL_NAME = "insert_test_tbl"
	FQ_TBL_NAME = unique_database + "." + TBL_NAME
	self.execute_query("create external table %s (i int) "
	"partitioned by (j int, s string)" % FQ_TBL_NAME)

	# initialize compute stats time
	self.execute_query("compute stats %s" % FQ_TBL_NAME)

	h = TestLastDdlTimeUpdate.TestHelper(self, unique_database, TBL_NAME)

	# static partition insert
	h.expect_no_time_change("insert into %(TBL)s partition(j=1, s='2012') select 10")
	# dynamic partition insert
	h.expect_no_time_change("insert into %(TBL)s partition(j, s) select 10, 2, '2013'")

	# dynamic partition insert changing no partitions (empty input)
	h.expect_no_time_change("insert into %(TBL)s partition(j, s) "
	"select * from (select 10 as i, 2 as j, '2013' as s) as t "
	"where t.i < 10")
	# dynamic partition insert modifying an existing partition
	h.expect_no_time_change("insert into %(TBL)s partition(j, s) select 20, 1, '2012'")

	def test_kudu(self, vector, unique_database):
	TBL_NAME = "kudu_test_tbl"
	FQ_TBL_NAME = unique_database + "." + TBL_NAME
	self.execute_query("create table %s (i int primary key) "
	"partition by hash(i) partitions 3 stored as kudu" % FQ_TBL_NAME)

	# initialize last compute stats time
	self.execute_query("compute stats %s" % FQ_TBL_NAME)

	h = TestLastDdlTimeUpdate.TestHelper(self, unique_database, TBL_NAME)

	# insert
	h.expect_no_time_change("insert into %s values (1)" % FQ_TBL_NAME)

	self.run_common_test_cases(h)

	# Tests that should behave the same with HDFS and Kudu tables.
	def run_common_test_cases(self, test_helper):
	h = test_helper
	# rename columns
	h.expect_ddl_time_change("alter table %(TBL)s change column i k int")
	h.expect_ddl_time_change("alter table %(TBL)s change column k i int")
	# change table property
	h.expect_ddl_time_change("alter table %(TBL)s set tblproperties ('a'='b')")

	# changing table statistics manually
	h.expect_ddl_time_change("alter table %(TBL)s set tblproperties ('numRows'='1')")

	# changing column statistics manually
	h.expect_no_time_change("alter table %(TBL)s set column stats i ('numdvs'='3')")

	# compute statistics
	h.expect_stat_time_change("compute stats %(TBL)s")
	# compute statistics for a single column
	h.expect_stat_time_change("compute stats %(TBL)s (i)")

	# drop statistics
	h.expect_no_time_change("drop stats %(TBL)s")

	# invalidate metadata and reload table
	self.execute_query("invalidate metadata %s" % h.fq_tbl_name)
	# run any query to reload the table
	h.expect_no_time_change("describe %(TBL)s")