| # 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 glob |
| import os |
| import pytest |
| import re |
| import shutil |
| import subprocess |
| |
| from tempfile import mkdtemp |
| from tests.common.custom_cluster_test_suite import CustomClusterTestSuite |
| from tests.common.skip import SkipIfS3, SkipIfABFS, SkipIfADLS, SkipIfIsilon, SkipIfLocal |
| from tests.common.test_dimensions import create_uncompressed_text_dimension |
| from tests.util.filesystem_utils import get_fs_path |
| |
| class TestUdfPersistence(CustomClusterTestSuite): |
| """ Tests the behavior of UDFs and UDAs between catalog restarts. With IMPALA-1748, |
| these functions are persisted to the metastore and are loaded again during catalog |
| startup""" |
| |
| DATABASE = 'udf_permanent_test' |
| JAVA_FN_TEST_DB = 'java_permanent_test' |
| HIVE_IMPALA_INTEGRATION_DB = 'hive_impala_integration_db' |
| HIVE_UDF_JAR = os.getenv('DEFAULT_FS') + '/test-warehouse/hive-exec.jar'; |
| JAVA_UDF_JAR = os.getenv('DEFAULT_FS') + '/test-warehouse/impala-hive-udfs.jar'; |
| LOCAL_LIBRARY_DIR = mkdtemp(dir="/tmp") |
| |
| @classmethod |
| def get_workload(cls): |
| return 'functional-query' |
| |
| @classmethod |
| def setup_class(cls): |
| if cls.exploration_strategy() != 'exhaustive': |
| pytest.skip('runs only in exhaustive') |
| super(TestUdfPersistence, cls).setup_class() |
| |
| @classmethod |
| def add_test_dimensions(cls): |
| super(TestUdfPersistence, cls).add_test_dimensions() |
| cls.ImpalaTestMatrix.add_dimension( |
| create_uncompressed_text_dimension(cls.get_workload())) |
| |
| def setup_method(self, method): |
| super(TestUdfPersistence, self).setup_method(method) |
| impalad = self.cluster.impalads[0] |
| self.client = impalad.service.create_beeswax_client() |
| self.__cleanup() |
| self.__load_drop_functions( |
| self.CREATE_UDFS_TEMPLATE, self.DATABASE, |
| get_fs_path('/test-warehouse/libTestUdfs.so')) |
| self.__load_drop_functions( |
| self.DROP_SAMPLE_UDAS_TEMPLATE, self.DATABASE, |
| get_fs_path('/test-warehouse/libudasample.so')) |
| self.__load_drop_functions( |
| self.CREATE_SAMPLE_UDAS_TEMPLATE, self.DATABASE, |
| get_fs_path('/test-warehouse/libudasample.so')) |
| self.__load_drop_functions( |
| self.CREATE_TEST_UDAS_TEMPLATE, self.DATABASE, |
| get_fs_path('/test-warehouse/libTestUdas.so')) |
| self.uda_count =\ |
| self.CREATE_SAMPLE_UDAS_TEMPLATE.count("create aggregate function") +\ |
| self.CREATE_TEST_UDAS_TEMPLATE.count("create aggregate function") |
| self.udf_count = self.CREATE_UDFS_TEMPLATE.count("create function") |
| self.client.execute("CREATE DATABASE IF NOT EXISTS %s" % self.JAVA_FN_TEST_DB) |
| self.client.execute("CREATE DATABASE IF NOT EXISTS %s" % |
| self.HIVE_IMPALA_INTEGRATION_DB) |
| |
| def teardown_method(self, method): |
| self.__cleanup() |
| |
| def __cleanup(self): |
| self.client.execute("DROP DATABASE IF EXISTS %s CASCADE" % self.DATABASE) |
| self.client.execute("DROP DATABASE IF EXISTS %s CASCADE" % self.JAVA_FN_TEST_DB) |
| self.client.execute("DROP DATABASE IF EXISTS %s CASCADE" |
| % self.HIVE_IMPALA_INTEGRATION_DB) |
| shutil.rmtree(self.LOCAL_LIBRARY_DIR, ignore_errors=True) |
| |
| def __load_drop_functions(self, template, database, location): |
| queries = template.format(database=database, location=location) |
| # Split queries and remove empty lines |
| queries = [q for q in queries.split(';') if q.strip()] |
| for query in queries: |
| result = self.client.execute(query) |
| assert result is not None |
| |
| def __restart_cluster(self): |
| self._stop_impala_cluster() |
| self._start_impala_cluster(list()) |
| impalad = self.cluster.impalads[0] |
| self.client = impalad.service.create_beeswax_client() |
| |
| def verify_function_count(self, query, count): |
| result = self.client.execute(query) |
| assert result is not None and len(result.data) == count |
| |
| @pytest.mark.execute_serially |
| def test_permanent_udfs(self): |
| # Make sure the pre-calculated count tallies with the number of |
| # functions shown using "show [aggregate] functions" statement |
| self.verify_function_count( |
| "SHOW FUNCTIONS in {0}".format(self.DATABASE), self.udf_count); |
| self.verify_function_count( |
| "SHOW AGGREGATE FUNCTIONS in {0}".format(self.DATABASE), self.uda_count) |
| # invalidate metadata and make sure the count tallies |
| result = self.client.execute("INVALIDATE METADATA") |
| self.verify_function_count( |
| "SHOW FUNCTIONS in {0}".format(self.DATABASE), self.udf_count); |
| self.verify_function_count( |
| "SHOW AGGREGATE FUNCTIONS in {0}".format(self.DATABASE), self.uda_count) |
| # Restart the cluster, this triggers a full metadata reload |
| self.__restart_cluster() |
| # Make sure the counts of udfs and udas match post restart |
| self.verify_function_count( |
| "SHOW FUNCTIONS in {0}".format(self.DATABASE), self.udf_count); |
| self.verify_function_count( |
| "SHOW AGGREGATE FUNCTIONS in {0}".format(self.DATABASE), self.uda_count) |
| # Drop sample udas and verify the count matches pre and post restart |
| self.__load_drop_functions( |
| self.DROP_SAMPLE_UDAS_TEMPLATE, self.DATABASE, |
| get_fs_path('/test-warehouse/libudasample.so')) |
| self.verify_function_count( |
| "SHOW AGGREGATE FUNCTIONS in {0}".format(self.DATABASE), 1) |
| self.__restart_cluster() |
| self.verify_function_count( |
| "SHOW AGGREGATE FUNCTIONS in {0}".format(self.DATABASE), 1) |
| |
| |
| def __verify_udf_in_hive(self, udf): |
| (query, result) = self.SAMPLE_JAVA_UDFS_TEST[udf] |
| stdout = self.run_stmt_in_hive("select " + query.format( |
| db=self.HIVE_IMPALA_INTEGRATION_DB)) |
| assert stdout is not None and result in str(stdout) |
| |
| def __verify_udf_in_impala(self, udf): |
| (query, result) = self.SAMPLE_JAVA_UDFS_TEST[udf] |
| stdout = self.client.execute("select " + query.format( |
| db=self.HIVE_IMPALA_INTEGRATION_DB)) |
| assert stdout is not None and result in str(stdout.data) |
| |
| def __describe_udf_in_hive(self, udf, db=HIVE_IMPALA_INTEGRATION_DB): |
| """ Describe the specified function, returning stdout. """ |
| # Hive 2+ caches UDFs, so we have to explicitly invalidate the UDF if |
| # we've made changes on the Impala side. |
| stmt = "RELOAD FUNCTION ; DESCRIBE FUNCTION {0}.{1}".format(db, udf) |
| return self.run_stmt_in_hive(stmt) |
| |
| @SkipIfIsilon.hive |
| @SkipIfS3.hive |
| @SkipIfABFS.hive |
| @SkipIfADLS.hive |
| @SkipIfLocal.hive |
| @pytest.mark.execute_serially |
| def test_corrupt_java_udf(self): |
| """ IMPALA-3820: This tests if the Catalog server can gracefully handle |
| Java UDFs with unresolved dependencies.""" |
| if self.exploration_strategy() != 'exhaustive': pytest.skip() |
| # Create a Java UDF with unresolved dependencies from Hive and |
| # restart the Catalog server. Catalog should ignore the |
| # function load. |
| self.run_stmt_in_hive("create function %s.corrupt_udf as \ |
| 'org.apache.impala.UnresolvedUdf' using jar '%s'" |
| % (self.JAVA_FN_TEST_DB, self.JAVA_UDF_JAR)) |
| self.__restart_cluster() |
| # Make sure the function count is 0 |
| self.verify_function_count( |
| "SHOW FUNCTIONS in {0}".format(self.JAVA_FN_TEST_DB), 0) |
| |
| |
| @SkipIfIsilon.hive |
| @SkipIfS3.hive |
| @SkipIfABFS.hive |
| @SkipIfADLS.hive |
| @SkipIfLocal.hive |
| @pytest.mark.execute_serially |
| @CustomClusterTestSuite.with_args( |
| catalogd_args= "--local_library_dir={0}".format(LOCAL_LIBRARY_DIR)) |
| def test_java_udfs_hive_integration(self): |
| ''' This test checks the integration between Hive and Impala on |
| CREATE FUNCTION and DROP FUNCTION statements for persistent Java UDFs. |
| The main objective of the test is to check the following four cases. |
| - Add Java UDFs from Impala and make sure they are visible in Hive |
| - Drop Java UDFs from Impala and make sure this reflects in Hive. |
| - Add Java UDFs from Hive and make sure they are visitble in Impala |
| - Drop Java UDFs from Hive and make sure this reflects in Impala |
| ''' |
| # Add Java UDFs from Impala and check if they are visible in Hive. |
| # Hive has bug that doesn't display the permanent function in show functions |
| # statement. So this test relies on describe function statement which prints |
| # a message if the function is not present. |
| udfs_to_test = list(self.SAMPLE_JAVA_UDFS) |
| if int(os.environ['IMPALA_HIVE_MAJOR_VERSION']) == 2: |
| udfs_to_test += self.SAMPLE_JAVA_UDFS_HIVE2_ONLY |
| for (fn, fn_symbol) in udfs_to_test: |
| self.client.execute(self.DROP_JAVA_UDF_TEMPLATE.format( |
| db=self.HIVE_IMPALA_INTEGRATION_DB, function=fn)) |
| self.client.execute(self.CREATE_JAVA_UDF_TEMPLATE.format( |
| db=self.HIVE_IMPALA_INTEGRATION_DB, function=fn, |
| location=self.HIVE_UDF_JAR, symbol=fn_symbol)) |
| hive_stdout = self.__describe_udf_in_hive(fn) |
| assert "does not exist" not in hive_stdout |
| self.__verify_udf_in_hive(fn) |
| # Drop the function from Impala and check if it reflects in Hive. |
| self.client.execute(self.DROP_JAVA_UDF_TEMPLATE.format( |
| db=self.HIVE_IMPALA_INTEGRATION_DB, function=fn)) |
| hive_stdout = self.__describe_udf_in_hive(fn) |
| assert "does not exist" in hive_stdout |
| |
| # Create the same set of functions from Hive and make sure they are visible |
| # in Impala. There are two ways to make functions visible in Impala: invalidate |
| # metadata and refresh functions <db>. |
| REFRESH_COMMANDS = ["INVALIDATE METADATA", |
| "REFRESH FUNCTIONS {0}".format(self.HIVE_IMPALA_INTEGRATION_DB)] |
| for refresh_command in REFRESH_COMMANDS: |
| for (fn, fn_symbol) in udfs_to_test: |
| self.run_stmt_in_hive(self.CREATE_HIVE_UDF_TEMPLATE.format( |
| db=self.HIVE_IMPALA_INTEGRATION_DB, function=fn, |
| location=self.HIVE_UDF_JAR, symbol=fn_symbol)) |
| self.client.execute(refresh_command) |
| for (fn, fn_symbol) in udfs_to_test: |
| result = self.client.execute("SHOW FUNCTIONS IN {0}".format( |
| self.HIVE_IMPALA_INTEGRATION_DB)) |
| assert result is not None and len(result.data) > 0 and\ |
| fn in str(result.data) |
| self.__verify_udf_in_impala(fn) |
| # Drop the function in Hive and make sure it reflects in Impala. |
| self.run_stmt_in_hive(self.DROP_JAVA_UDF_TEMPLATE.format( |
| db=self.HIVE_IMPALA_INTEGRATION_DB, function=fn)) |
| self.client.execute(refresh_command) |
| self.verify_function_count( |
| "SHOW FUNCTIONS in {0}".format(self.HIVE_IMPALA_INTEGRATION_DB), 0) |
| # Make sure we deleted all the temporary jars we copied to the local fs |
| assert len(glob.glob(self.LOCAL_LIBRARY_DIR + "/*.jar")) == 0 |
| |
| @SkipIfIsilon.hive |
| @SkipIfS3.hive |
| @SkipIfABFS.hive |
| @SkipIfADLS.hive |
| @SkipIfLocal.hive |
| @pytest.mark.execute_serially |
| @CustomClusterTestSuite.with_args( |
| catalogd_args= "--local_library_dir={0}".format(LOCAL_LIBRARY_DIR)) |
| def test_refresh_native(self): |
| ''' This test checks that a native function is visible in Impala after a |
| REFRESH FUNCTIONS command. We will add the native function through Hive |
| by setting DBPROPERTIES of a database.''' |
| # First we create the function in Impala. |
| create_func_impala = ("create function {database}.identity_tmp(bigint) " |
| "returns bigint location '{location}' symbol='Identity'") |
| self.client.execute(create_func_impala.format( |
| database=self.HIVE_IMPALA_INTEGRATION_DB, |
| location=get_fs_path('/test-warehouse/libTestUdfs.so'))) |
| |
| # Impala puts the native function into a database property table. We extract the key |
| # value pair that represents the function from the table. |
| describe_db_hive = "DESCRIBE DATABASE EXTENDED {database}".format( |
| database=self.HIVE_IMPALA_INTEGRATION_DB) |
| result = self.run_stmt_in_hive(describe_db_hive) |
| regex = r"{(.*?)=(.*?)}" |
| match = re.search(regex, result) |
| func_name = match.group(1) |
| func_contents = match.group(2) |
| |
| # Recreate the database, this deletes the function. |
| self.client.execute("DROP DATABASE {database} CASCADE".format( |
| database=self.HIVE_IMPALA_INTEGRATION_DB)) |
| self.client.execute("CREATE DATABASE {database}".format( |
| database=self.HIVE_IMPALA_INTEGRATION_DB)) |
| result = self.client.execute("SHOW FUNCTIONS IN {database}".format( |
| database=self.HIVE_IMPALA_INTEGRATION_DB)) |
| assert result is not None and len(result.data) == 0 |
| |
| # Place the function into the recreated database by modifying it's properties. |
| alter_db_hive = "ALTER DATABASE {database} SET DBPROPERTIES ('{fn_name}'='{fn_val}')" |
| self.run_stmt_in_hive(alter_db_hive.format( |
| database=self.HIVE_IMPALA_INTEGRATION_DB, |
| fn_name=func_name, |
| fn_val=func_contents)) |
| result = self.client.execute("SHOW FUNCTIONS IN {database}".format( |
| database=self.HIVE_IMPALA_INTEGRATION_DB)) |
| assert result is not None and len(result.data) == 0 |
| |
| # The function should be visible in Impala after a REFRESH FUNCTIONS. |
| self.client.execute("REFRESH FUNCTIONS {database}".format( |
| database=self.HIVE_IMPALA_INTEGRATION_DB)) |
| result = self.client.execute("SHOW FUNCTIONS IN {database}".format( |
| database=self.HIVE_IMPALA_INTEGRATION_DB)) |
| assert result is not None and len(result.data) > 0 and\ |
| "identity_tmp" in str(result.data) |
| |
| # Verify that the function returns a correct result. |
| result = self.client.execute("SELECT {database}.identity_tmp(10)".format( |
| database=self.HIVE_IMPALA_INTEGRATION_DB)) |
| assert result.data[0] == "10" |
| # Make sure we deleted all the temporary jars we copied to the local fs |
| assert len(glob.glob(self.LOCAL_LIBRARY_DIR + "/*.jar")) == 0 |
| |
| @SkipIfIsilon.hive |
| @SkipIfS3.hive |
| @SkipIfABFS.hive |
| @SkipIfADLS.hive |
| @SkipIfLocal.hive |
| @pytest.mark.execute_serially |
| @CustomClusterTestSuite.with_args( |
| catalogd_args= "--local_library_dir={0}".format(LOCAL_LIBRARY_DIR)) |
| def test_refresh_replace(self): |
| ''' This test checks that if we drop a function and then create a |
| different function with the same name in Hive, the new function will |
| be visible in Impala after REFRESH FUNCTIONS.''' |
| # Create an original function. |
| create_orig_func_hive = ("create function {database}.test_func as " |
| "'org.apache.hadoop.hive.ql.udf.UDFHex' using jar '{jar}'") |
| self.run_stmt_in_hive(create_orig_func_hive.format( |
| database=self.HIVE_IMPALA_INTEGRATION_DB, jar=self.JAVA_UDF_JAR)) |
| result = self.client.execute("SHOW FUNCTIONS IN {database}".format( |
| database=self.HIVE_IMPALA_INTEGRATION_DB)) |
| assert result is not None and len(result.data) == 0 |
| # Verify the function becomes visible in Impala after REFRESH FUNCTIONS. |
| self.client.execute("REFRESH FUNCTIONS {database}".format( |
| database=self.HIVE_IMPALA_INTEGRATION_DB)) |
| result = self.client.execute("SHOW FUNCTIONS IN {database}".format( |
| database=self.HIVE_IMPALA_INTEGRATION_DB)) |
| assert (result is not None and len(result.data) == 3 and |
| "test_func" in str(result.data)) |
| result = self.client.execute("SELECT {database}.test_func(123)".format( |
| database=self.HIVE_IMPALA_INTEGRATION_DB)) |
| assert result.data[0] == "7B" |
| |
| # Drop the original function and create a different function with the same name as |
| # the original, but a different JAR. |
| drop_orig_func_hive = "DROP FUNCTION {database}.test_func" |
| self.run_stmt_in_hive(drop_orig_func_hive.format( |
| database=self.HIVE_IMPALA_INTEGRATION_DB)) |
| create_replacement_func_hive = ("create function {database}.test_func as " |
| "'org.apache.hadoop.hive.ql.udf.UDFBin' using jar '{jar}'") |
| self.run_stmt_in_hive(create_replacement_func_hive.format( |
| database=self.HIVE_IMPALA_INTEGRATION_DB, jar=self.JAVA_UDF_JAR)) |
| self.client.execute("REFRESH FUNCTIONS {database}".format( |
| database=self.HIVE_IMPALA_INTEGRATION_DB)) |
| result = self.client.execute("SHOW FUNCTIONS IN {database}".format( |
| database=self.HIVE_IMPALA_INTEGRATION_DB)) |
| assert (result is not None and len(result.data) == 1 and |
| "test_func" in str(result.data)) |
| # Verify that the function has actually been updated. |
| result = self.client.execute("SELECT {database}.test_func(123)".format( |
| database=self.HIVE_IMPALA_INTEGRATION_DB)) |
| assert result.data[0] == "1111011" |
| # Make sure we deleted all the temporary jars we copied to the local fs |
| assert len(glob.glob(self.LOCAL_LIBRARY_DIR + "/*.jar")) == 0 |
| |
| @pytest.mark.execute_serially |
| def test_java_udfs_from_impala(self): |
| """ This tests checks the behavior of permanent Java UDFs in Impala.""" |
| self.verify_function_count( |
| "SHOW FUNCTIONS in {0}".format(self.JAVA_FN_TEST_DB), 0); |
| # Create a non persistent Java UDF and make sure we can't create a |
| # persistent Java UDF with same name |
| self.client.execute("create function %s.%s(boolean) returns boolean "\ |
| "location '%s' symbol='%s'" % (self.JAVA_FN_TEST_DB, "identity", |
| self.JAVA_UDF_JAR, "org.apache.impala.TestUdf")) |
| result = self.execute_query_expect_failure(self.client, |
| self.CREATE_JAVA_UDF_TEMPLATE.format(db=self.JAVA_FN_TEST_DB, |
| function="identity", location=self.JAVA_UDF_JAR, |
| symbol="org.apache.impala.TestUdf")) |
| assert "Function already exists" in str(result) |
| # Test the same with a NATIVE function |
| self.client.execute("create function {database}.identity(int) "\ |
| "returns int location '{location}' symbol='Identity'".format( |
| database=self.JAVA_FN_TEST_DB, |
| location="/test-warehouse/libTestUdfs.so")) |
| result = self.execute_query_expect_failure(self.client, |
| self.CREATE_JAVA_UDF_TEMPLATE.format(db=self.JAVA_FN_TEST_DB, |
| function="identity", location=self.JAVA_UDF_JAR, |
| symbol="org.apache.impala.TestUdf")) |
| assert "Function already exists" in str(result) |
| |
| # Test the reverse. Add a persistent Java UDF and ensure we cannot |
| # add non persistent Java UDFs or NATIVE functions with the same name. |
| self.client.execute(self.CREATE_JAVA_UDF_TEMPLATE.format( |
| db=self.JAVA_FN_TEST_DB, function="identity_java", |
| location=self.JAVA_UDF_JAR, symbol="org.apache.impala.TestUdf")) |
| result = self.execute_query_expect_failure(self.client, "create function "\ |
| "%s.%s(boolean) returns boolean location '%s' symbol='%s'" % ( |
| self.JAVA_FN_TEST_DB, "identity_java", self.JAVA_UDF_JAR, |
| "org.apache.impala.TestUdf")) |
| assert "Function already exists" in str(result) |
| result = self.execute_query_expect_failure(self.client, "create function "\ |
| "{database}.identity_java(int) returns int location '{location}' "\ |
| "symbol='Identity'".format(database=self.JAVA_FN_TEST_DB, |
| location="/test-warehouse/libTestUdfs.so")) |
| assert "Function already exists" in str(result) |
| # With IF NOT EXISTS, the query shouldn't fail. |
| result = self.execute_query_expect_success(self.client, "create function "\ |
| " if not exists {database}.identity_java(int) returns int location "\ |
| "'{location}' symbol='Identity'".format(database=self.JAVA_FN_TEST_DB, |
| location="/test-warehouse/libTestUdfs.so")) |
| result = self.client.execute("SHOW FUNCTIONS in %s" % self.JAVA_FN_TEST_DB) |
| self.execute_query_expect_success(self.client, |
| "DROP FUNCTION IF EXISTS {db}.impala_java".format(db=self.JAVA_FN_TEST_DB)) |
| |
| # Drop the persistent Java function. |
| # Test the same create with IF NOT EXISTS. No exception should be thrown. |
| # Add a Java udf which has a few incompatible 'evaluate' functions in the |
| # symbol class. Catalog should load only the compatible ones. JavaUdfTest |
| # has 8 evaluate signatures out of which only 3 are valid. |
| compatibility_fn_count = 3 |
| self.client.execute(self.CREATE_JAVA_UDF_TEMPLATE.format( |
| db=self.JAVA_FN_TEST_DB, function="compatibility", |
| location=self.JAVA_UDF_JAR, symbol="org.apache.impala.JavaUdfTest")) |
| self.verify_function_count( |
| "SHOW FUNCTIONS IN %s like 'compatibility*'" % self.JAVA_FN_TEST_DB, |
| compatibility_fn_count) |
| result = self.client.execute("SHOW FUNCTIONS in %s" % self.JAVA_FN_TEST_DB) |
| function_count = len(result.data) |
| # Invalidating metadata should preserve all the functions |
| self.client.execute("INVALIDATE METADATA") |
| self.verify_function_count( |
| "SHOW FUNCTIONS IN %s" % self.JAVA_FN_TEST_DB, function_count) |
| # Restarting the cluster should preserve only the persisted functions. In |
| # this case, identity(boolean) should be wiped out. |
| self.__restart_cluster() |
| self.verify_function_count( |
| "SHOW FUNCTIONS IN %s" % self.JAVA_FN_TEST_DB, function_count-1) |
| # Dropping persisted Java UDFs with old syntax should raise an exception |
| self.execute_query_expect_failure(self.client, |
| "DROP FUNCTION compatibility(smallint)") |
| self.verify_function_count( |
| "SHOW FUNCTIONS IN %s like 'compatibility*'" % self.JAVA_FN_TEST_DB, 3) |
| # Drop the functions and make sure they don't appear post restart. |
| self.client.execute("DROP FUNCTION %s.compatibility" % self.JAVA_FN_TEST_DB) |
| self.verify_function_count( |
| "SHOW FUNCTIONS IN %s like 'compatibility*'" % self.JAVA_FN_TEST_DB, 0) |
| self.__restart_cluster() |
| self.verify_function_count( |
| "SHOW FUNCTIONS IN %s like 'compatibility*'" % self.JAVA_FN_TEST_DB, 0) |
| |
| # Try to load a UDF that has no compatible signatures. Make sure it is not added |
| # to Hive and Impala. |
| result = self.execute_query_expect_failure(self.client, |
| self.CREATE_JAVA_UDF_TEMPLATE.format(db=self.JAVA_FN_TEST_DB, function="badudf", |
| location=self.JAVA_UDF_JAR, symbol="org.apache.impala.IncompatibleUdfTest")) |
| assert "No compatible function signatures" in str(result) |
| self.verify_function_count( |
| "SHOW FUNCTIONS IN %s like 'badudf*'" % self.JAVA_FN_TEST_DB, 0) |
| result = self.__describe_udf_in_hive('badudf', db=self.JAVA_FN_TEST_DB) |
| assert "does not exist" in str(result) |
| # Create the same function from hive and make sure Impala doesn't load any signatures. |
| self.run_stmt_in_hive(self.CREATE_HIVE_UDF_TEMPLATE.format( |
| db=self.JAVA_FN_TEST_DB, function="badudf", |
| location=self.JAVA_UDF_JAR, symbol="org.apache.impala.IncompatibleUdfTest")) |
| result = self.__describe_udf_in_hive('badudf', db=self.JAVA_FN_TEST_DB) |
| assert "does not exist" not in str(result) |
| self.client.execute("INVALIDATE METADATA") |
| self.verify_function_count( |
| "SHOW FUNCTIONS IN %s like 'badudf*'" % self.JAVA_FN_TEST_DB, 0) |
| # Add a function with the same name from Impala. It should fail. |
| result = self.execute_query_expect_failure(self.client, |
| self.CREATE_JAVA_UDF_TEMPLATE.format(db=self.JAVA_FN_TEST_DB, function="badudf", |
| location=self.JAVA_UDF_JAR, symbol="org.apache.impala.TestUdf")) |
| assert "Function badudf already exists" in str(result) |
| # Drop the function and make sure the function if dropped from hive |
| self.client.execute(self.DROP_JAVA_UDF_TEMPLATE.format( |
| db=self.JAVA_FN_TEST_DB, function="badudf")) |
| result = self.__describe_udf_in_hive('badudf', db=self.JAVA_FN_TEST_DB) |
| assert "does not exist" in str(result) |
| |
| # Create sample UDA functions in {database} from library {location} |
| |
| DROP_SAMPLE_UDAS_TEMPLATE = """ |
| drop function if exists {database}.test_count(int); |
| drop function if exists {database}.hll(int); |
| drop function if exists {database}.sum_small_decimal(decimal(9,2)); |
| """ |
| |
| CREATE_JAVA_UDF_TEMPLATE = """ |
| CREATE FUNCTION {db}.{function} LOCATION '{location}' symbol='{symbol}' |
| """ |
| |
| CREATE_HIVE_UDF_TEMPLATE = """ |
| CREATE FUNCTION {db}.{function} as '{symbol}' USING JAR '{location}' |
| """ |
| |
| DROP_JAVA_UDF_TEMPLATE = "DROP FUNCTION IF EXISTS {db}.{function}" |
| |
| # Sample java udfs from hive-exec.jar. Function name to symbol class mapping |
| SAMPLE_JAVA_UDFS = [ |
| ('udfpi', 'org.apache.hadoop.hive.ql.udf.UDFPI'), |
| ('udfbin', 'org.apache.hadoop.hive.ql.udf.UDFBin'), |
| ('udfhex', 'org.apache.hadoop.hive.ql.udf.UDFHex'), |
| ('udfconv', 'org.apache.hadoop.hive.ql.udf.UDFConv'), |
| ('udflike', 'org.apache.hadoop.hive.ql.udf.UDFLike'), |
| ('udfsign', 'org.apache.hadoop.hive.ql.udf.UDFSign'), |
| ('udfascii','org.apache.hadoop.hive.ql.udf.UDFAscii') |
| ] |
| |
| # These UDFs are available in Hive 2 but in Hive 3 are now implemented |
| # using a new GenericUDF interface that we don't support. |
| SAMPLE_JAVA_UDFS_HIVE2_ONLY = [ |
| ('udfhour', 'org.apache.hadoop.hive.ql.udf.UDFHour'), |
| ('udfyear', 'org.apache.hadoop.hive.ql.udf.UDFYear'), |
| ] |
| |
| # Simple tests to verify java udfs in SAMPLE_JAVA_UDFS |
| SAMPLE_JAVA_UDFS_TEST = { |
| 'udfpi' : ('{db}.udfpi()', '3.141592653589793'), |
| 'udfbin' : ('{db}.udfbin(123)', '1111011'), |
| 'udfhex' : ('{db}.udfhex(123)', '7B'), |
| 'udfconv' : ('{db}.udfconv("100", 2, 10)', '4'), |
| 'udfhour' : ('{db}.udfhour("12:55:12")', '12'), |
| 'udflike' : ('{db}.udflike("abc", "def")', 'false'), |
| 'udfsign' : ('{db}.udfsign(0)', '0'), |
| 'udfyear' : ('{db}.udfyear("1990-02-06")', '1990'), |
| 'udfascii' : ('{db}.udfascii("abc")','97') |
| } |
| |
| CREATE_SAMPLE_UDAS_TEMPLATE = """ |
| create database if not exists {database}; |
| |
| create aggregate function {database}.test_count(int) returns bigint |
| location '{location}' update_fn='CountUpdate'; |
| |
| create aggregate function {database}.hll(int) returns string |
| location '{location}' update_fn='HllUpdate'; |
| |
| create aggregate function {database}.sum_small_decimal(decimal(9,2)) |
| returns decimal(9,2) location '{location}' update_fn='SumSmallDecimalUpdate'; |
| """ |
| |
| # Create test UDA functions in {database} from library {location} |
| CREATE_TEST_UDAS_TEMPLATE = """ |
| drop function if exists {database}.trunc_sum(double); |
| |
| create database if not exists {database}; |
| |
| create aggregate function {database}.trunc_sum(double) |
| returns bigint intermediate double location '{location}' |
| update_fn='TruncSumUpdate' merge_fn='TruncSumMerge' |
| serialize_fn='TruncSumSerialize' finalize_fn='TruncSumFinalize'; |
| """ |
| |
| # Create test UDF functions in {database} from library {location} |
| CREATE_UDFS_TEMPLATE = """ |
| drop function if exists {database}.identity(boolean); |
| drop function if exists {database}.identity(tinyint); |
| drop function if exists {database}.identity(smallint); |
| drop function if exists {database}.identity(int); |
| drop function if exists {database}.identity(bigint); |
| drop function if exists {database}.identity(float); |
| drop function if exists {database}.identity(double); |
| drop function if exists {database}.identity(string); |
| drop function if exists {database}.identity(timestamp); |
| drop function if exists {database}.identity(date); |
| drop function if exists {database}.identity(decimal(9,0)); |
| drop function if exists {database}.identity(decimal(18,1)); |
| drop function if exists {database}.identity(decimal(38,10)); |
| drop function if exists {database}.all_types_fn( |
| string, boolean, tinyint, smallint, int, bigint, float, double, decimal(2,0), |
| date); |
| drop function if exists {database}.no_args(); |
| drop function if exists {database}.var_and(boolean...); |
| drop function if exists {database}.var_sum(int...); |
| drop function if exists {database}.var_sum(double...); |
| drop function if exists {database}.var_sum(string...); |
| drop function if exists {database}.var_sum(decimal(4,2)...); |
| drop function if exists {database}.var_sum_multiply(double, int...); |
| drop function if exists {database}.constant_timestamp(); |
| drop function if exists {database}.constant_date(); |
| drop function if exists {database}.validate_arg_type(string); |
| drop function if exists {database}.count_rows(); |
| drop function if exists {database}.constant_arg(int); |
| drop function if exists {database}.validate_open(int); |
| drop function if exists {database}.mem_test(bigint); |
| drop function if exists {database}.mem_test_leaks(bigint); |
| drop function if exists {database}.unmangled_symbol(); |
| drop function if exists {database}.four_args(int, int, int, int); |
| drop function if exists {database}.five_args(int, int, int, int, int); |
| drop function if exists {database}.six_args(int, int, int, int, int, int); |
| drop function if exists {database}.seven_args(int, int, int, int, int, int, int); |
| drop function if exists {database}.eight_args(int, int, int, int, int, int, int, int); |
| |
| create database if not exists {database}; |
| |
| create function {database}.identity(boolean) returns boolean |
| location '{location}' symbol='Identity'; |
| |
| create function {database}.identity(tinyint) returns tinyint |
| location '{location}' symbol='Identity'; |
| |
| create function {database}.identity(smallint) returns smallint |
| location '{location}' symbol='Identity'; |
| |
| create function {database}.identity(int) returns int |
| location '{location}' symbol='Identity'; |
| |
| create function {database}.identity(bigint) returns bigint |
| location '{location}' symbol='Identity'; |
| |
| create function {database}.identity(float) returns float |
| location '{location}' symbol='Identity'; |
| |
| create function {database}.identity(double) returns double |
| location '{location}' symbol='Identity'; |
| |
| create function {database}.identity(string) returns string |
| location '{location}' |
| symbol='_Z8IdentityPN10impala_udf15FunctionContextERKNS_9StringValE'; |
| |
| create function {database}.identity(timestamp) returns timestamp |
| location '{location}' |
| symbol='_Z8IdentityPN10impala_udf15FunctionContextERKNS_12TimestampValE'; |
| |
| create function {database}.identity(date) returns date |
| location '{location}' |
| symbol='_Z8IdentityPN10impala_udf15FunctionContextERKNS_7DateValE'; |
| |
| create function {database}.identity(decimal(9,0)) returns decimal(9,0) |
| location '{location}' |
| symbol='_Z8IdentityPN10impala_udf15FunctionContextERKNS_10DecimalValE'; |
| |
| create function {database}.identity(decimal(18,1)) returns decimal(18,1) |
| location '{location}' |
| symbol='_Z8IdentityPN10impala_udf15FunctionContextERKNS_10DecimalValE'; |
| |
| create function {database}.identity(decimal(38,10)) returns decimal(38,10) |
| location '{location}' |
| symbol='_Z8IdentityPN10impala_udf15FunctionContextERKNS_10DecimalValE'; |
| |
| create function {database}.all_types_fn( |
| string, boolean, tinyint, smallint, int, bigint, float, double, decimal(2,0), |
| date) |
| returns int |
| location '{location}' symbol='AllTypes'; |
| |
| create function {database}.no_args() returns string |
| location '{location}' |
| symbol='_Z6NoArgsPN10impala_udf15FunctionContextE'; |
| |
| create function {database}.var_and(boolean...) returns boolean |
| location '{location}' symbol='VarAnd'; |
| |
| create function {database}.var_sum(int...) returns int |
| location '{location}' symbol='VarSum'; |
| |
| create function {database}.var_sum(double...) returns double |
| location '{location}' symbol='VarSum'; |
| |
| create function {database}.var_sum(string...) returns int |
| location '{location}' symbol='VarSum'; |
| |
| create function {database}.var_sum(decimal(4,2)...) returns decimal(18,2) |
| location '{location}' symbol='VarSum'; |
| |
| create function {database}.var_sum_multiply(double, int...) returns double |
| location '{location}' |
| symbol='_Z14VarSumMultiplyPN10impala_udf15FunctionContextERKNS_9DoubleValEiPKNS_6IntValE'; |
| |
| create function {database}.constant_timestamp() returns timestamp |
| location '{location}' symbol='ConstantTimestamp'; |
| |
| create function {database}.constant_date() returns date |
| location '{location}' symbol='ConstantDate'; |
| |
| create function {database}.validate_arg_type(string) returns boolean |
| location '{location}' symbol='ValidateArgType'; |
| |
| create function {database}.count_rows() returns bigint |
| location '{location}' symbol='Count' prepare_fn='CountPrepare' close_fn='CountClose'; |
| |
| create function {database}.constant_arg(int) returns int |
| location '{location}' symbol='ConstantArg' prepare_fn='ConstantArgPrepare' close_fn='ConstantArgClose'; |
| |
| create function {database}.validate_open(int) returns boolean |
| location '{location}' symbol='ValidateOpen' |
| prepare_fn='ValidateOpenPrepare' close_fn='ValidateOpenClose'; |
| |
| create function {database}.mem_test(bigint) returns bigint |
| location '{location}' symbol='MemTest' |
| prepare_fn='MemTestPrepare' close_fn='MemTestClose'; |
| |
| create function {database}.mem_test_leaks(bigint) returns bigint |
| location '{location}' symbol='MemTest' |
| prepare_fn='MemTestPrepare'; |
| """ |
