jprdonnelly/impala
1
0
Fork 0
mirror of https://github.com/apache/impala.git synced 2026-02-03 09:00:39 -05:00
Code Issues Projects Releases Wiki Activity
Files
c2ef633c60aff6ce3bf0fcffb6fc0763a7fdd3ca
impala/tests/query_test/test_insert.py
Peter Rozsa 3247cc6839 IMPALA-10173: Allow implicit casts between numeric and string types when inserting into table 
This patch adds an expiremental query option called ALLOW_UNSAFE_CASTS
which allows implicit casting between some numeric types and string
types. A new type of compatibility is introduced for this purpose, and
the compatibility rule handling is refactored also. The new approach
uses an enum to differentiate the compatibility levels, and to make it
easier to pass them through methods. The unsafe compatibility is used
only in two cases: for set operations and for insert statements. The
insert statements and set operations accept unsafe implicitly casted
expressions only when the source expressions are constant.

The following implicit type casts are enabled in unsafe mode:
  - String -> Float, Double
  - String -> Tinyint, Smallint, Int, Bigint
  - Float, Double -> String
  - Tinyint, Smallint, Int, Bigint -> String

The patch also covers IMPALA-3217, and adds two more rules to handle
implicit casting in set operations and insert statements between string
types:
  - String -> Char(n)
  - String -> Varchar(n)
The unsafe implicit casting requires that the source expression must be
constant in this case as well.

Tests:
  - tests added to AnalyzeExprsTest.java
  - new test class added to test_insert.py

Change-Id: Iee5db2301216c2e088b4b3e4f6cb5a1fd10600f7
Reviewed-on: http://gerrit.cloudera.org:8080/19881
Reviewed-by: Daniel Becker <daniel.becker@cloudera.com>
Tested-by: Impala Public Jenkins <impala-public-jenkins@cloudera.com>
2023-06-29 18:45:38 +00:00

561 lines
26 KiB
Python
Raw Blame History

# 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.
# Targeted Impala insert tests
from __future__ import absolute_import, division, print_function
import os
import pytest
import re
from testdata.common import widetable
from tests.common.impala_cluster import ImpalaCluster
from tests.common.impala_test_suite import ImpalaTestSuite
from tests.common.parametrize import UniqueDatabase
from tests.common.skip import (SkipIfFS, SkipIfLocal, SkipIfHive2,
SkipIfNotHdfsMinicluster)
from tests.common.test_dimensions import (
create_exec_option_dimension,
create_uncompressed_text_dimension,
create_single_exec_option_dimension,
is_supported_insert_format)
from tests.common.test_result_verifier import (
QueryTestResult,
parse_result_rows)
from tests.common.test_vector import ImpalaTestDimension
from tests.verifiers.metric_verifier import MetricVerifier
PARQUET_CODECS = ['none', 'snappy', 'gzip', 'zstd', 'lz4']
class TestInsertQueries(ImpalaTestSuite):
@classmethod
def get_workload(self):
return 'functional-query'
@classmethod
def add_test_dimensions(cls):
super(TestInsertQueries, cls).add_test_dimensions()
# Fix the exec_option vector to have a single value. This is needed should we decide
# to run the insert tests in parallel (otherwise there will be two tests inserting
# into the same table at the same time for the same file format).
# TODO: When we do decide to run these tests in parallel we could create unique temp
# tables for each test case to resolve the concurrency problems.
if cls.exploration_strategy() == 'core':
cls.ImpalaTestMatrix.add_dimension(create_exec_option_dimension(
cluster_sizes=[0], disable_codegen_options=[True, False], batch_sizes=[0],
sync_ddl=[0]))
cls.ImpalaTestMatrix.add_dimension(
create_uncompressed_text_dimension(cls.get_workload()))
else:
cls.ImpalaTestMatrix.add_dimension(create_exec_option_dimension(
cluster_sizes=[0], disable_codegen_options=[True, False], batch_sizes=[0, 1, 16],
sync_ddl=[0, 1]))
cls.ImpalaTestMatrix.add_dimension(
ImpalaTestDimension("compression_codec", *PARQUET_CODECS));
# Insert is currently only supported for text and parquet
# For parquet, we want to iterate through all the compression codecs
# TODO: each column in parquet can have a different codec. We could
# test all the codecs in one table/file with some additional flags.
cls.ImpalaTestMatrix.add_constraint(lambda v:\
v.get_value('table_format').file_format == 'parquet' or \
(v.get_value('table_format').file_format == 'text' and \
v.get_value('compression_codec') == 'none'))
cls.ImpalaTestMatrix.add_constraint(lambda v:\
v.get_value('table_format').compression_codec == 'none')
# Only test other batch sizes for uncompressed parquet to keep the execution time
# within reasonable bounds.
cls.ImpalaTestMatrix.add_constraint(lambda v:\
v.get_value('exec_option')['batch_size'] == 0 or \
(v.get_value('table_format').file_format == 'parquet' and \
v.get_value('compression_codec') == 'none'))
@pytest.mark.execute_serially
def test_insert_large_string(self, vector, unique_database):
"""Test handling of large strings in inserter and scanner."""
if "-Xcheck:jni" in os.environ.get("LIBHDFS_OPTS", ""):
pytest.skip("Test unreasonably slow with JNI checking.")
table_name = unique_database + ".insert_largestring"
self.client.set_configuration_option("mem_limit", "4gb")
self.client.set_configuration_option("max_row_size", "257mb")
file_format = vector.get_value('table_format').file_format
if file_format == "parquet":
stored_as = file_format
else:
assert file_format == "text"
stored_as = "textfile"
self.client.execute("""
create table {0}
stored as {1} as
select repeat('AZ', 128 * 1024 * 1024) as s""".format(table_name, stored_as))
# Make sure it produces correct result when materializing no tuples.
result = self.client.execute("select count(*) from {0}".format(table_name))
assert result.data == ["1"]
# Make sure it got the length right.
result = self.client.execute("select length(s) from {0}".format(table_name))
assert result.data == [str(2 * 128 * 1024 * 1024)]
# Spot-check the data.
result = self.client.execute(
"select substr(s, 200 * 1024 * 1024, 5) from {0}".format(table_name))
assert result.data == ["ZAZAZ"]
# IMPALA-7648: test that we gracefully fail when there is not enough memory
# to fit the scanned string in memory.
# IMPALA-9856: Disable result spooling for this query since it is intended to test
# for OOM.
self.client.set_configuration_option("spool_query_results", "0")
self.client.set_configuration_option("mem_limit", "50M")
try:
self.client.execute("select s from {0}".format(table_name))
assert False, "Expected query to fail"
except Exception as e:
assert "Memory limit exceeded" in str(e)
@classmethod
def setup_class(cls):
super(TestInsertQueries, cls).setup_class()
@UniqueDatabase.parametrize(sync_ddl=True)
# ABFS partition names cannot end in periods
@SkipIfFS.file_or_folder_name_ends_with_period
def test_insert(self, vector, unique_database):
if (vector.get_value('table_format').file_format == 'parquet'):
vector.get_value('exec_option')['COMPRESSION_CODEC'] = \
vector.get_value('compression_codec')
self.run_test_case('QueryTest/insert', vector, unique_database,
multiple_impalad=vector.get_value('exec_option')['sync_ddl'] == 1,
test_file_vars={'$ORIGINAL_DB': ImpalaTestSuite
.get_db_name_from_format(vector.get_value('table_format'))})
@SkipIfHive2.acid
@UniqueDatabase.parametrize(sync_ddl=True)
def test_acid_insert(self, vector, unique_database):
exec_options = vector.get_value('exec_option')
file_format = vector.get_value('table_format').file_format
if (file_format == 'parquet'):
exec_options['COMPRESSION_CODEC'] = vector.get_value('compression_codec')
exec_options['DEFAULT_FILE_FORMAT'] = file_format
self.run_test_case('QueryTest/acid-insert', vector, unique_database,
multiple_impalad=exec_options['sync_ddl'] == 1)
@SkipIfHive2.acid
@UniqueDatabase.parametrize(sync_ddl=True)
def test_acid_nonacid_insert(self, vector, unique_database):
self.run_test_case('QueryTest/acid-nonacid-insert', vector, unique_database,
multiple_impalad=vector.get_value('exec_option')['sync_ddl'] == 1)
@SkipIfHive2.acid
@UniqueDatabase.parametrize(sync_ddl=True)
def test_acid_insert_fail(self, vector, unique_database):
self.run_test_case('QueryTest/acid-insert-fail', vector, unique_database,
multiple_impalad=vector.get_value('exec_option')['sync_ddl'] == 1)
@UniqueDatabase.parametrize(sync_ddl=True)
@pytest.mark.execute_serially
@SkipIfNotHdfsMinicluster.tuned_for_minicluster
def test_insert_mem_limit(self, vector, unique_database):
if (vector.get_value('table_format').file_format == 'parquet'):
vector.get_value('exec_option')['COMPRESSION_CODEC'] = \
vector.get_value('compression_codec')
self.run_test_case('QueryTest/insert-mem-limit', vector, unique_database,
multiple_impalad=vector.get_value('exec_option')['sync_ddl'] == 1,
test_file_vars={'$ORIGINAL_DB': ImpalaTestSuite
.get_db_name_from_format(vector.get_value('table_format'))})
# IMPALA-7023: These queries can linger and use up memory, causing subsequent
# tests to hit memory limits. Wait for some time to allow the query to
# be reclaimed.
verifiers = [MetricVerifier(i.service)
for i in ImpalaCluster.get_e2e_test_cluster().impalads]
for v in verifiers:
v.wait_for_metric("impala-server.num-fragments-in-flight", 0, timeout=180)
@UniqueDatabase.parametrize(sync_ddl=True)
def test_insert_overwrite(self, vector, unique_database):
self.run_test_case('QueryTest/insert_overwrite', vector, unique_database,
multiple_impalad=vector.get_value('exec_option')['sync_ddl'] == 1,
test_file_vars={'$ORIGINAL_DB': ImpalaTestSuite
.get_db_name_from_format(vector.get_value('table_format'))})
@UniqueDatabase.parametrize(sync_ddl=True)
def test_insert_bad_expr(self, vector, unique_database):
# The test currently relies on codegen being disabled to trigger an error in
# the output expression of the table sink.
if vector.get_value('exec_option')['disable_codegen']:
self.run_test_case('QueryTest/insert_bad_expr', vector, unique_database,
multiple_impalad=vector.get_value('exec_option')['sync_ddl'] == 1,
test_file_vars={'$ORIGINAL_DB': ImpalaTestSuite
.get_db_name_from_format(vector.get_value('table_format'))})
@UniqueDatabase.parametrize(sync_ddl=True)
def test_insert_random_partition(self, vector, unique_database):
"""Regression test for IMPALA-402: partitioning by rand() leads to strange behaviour
or crashes."""
self.run_test_case('QueryTest/insert-random-partition', vector, unique_database,
multiple_impalad=vector.get_value('exec_option')['sync_ddl'] == 1)
class TestInsertWideTable(ImpalaTestSuite):
@classmethod
def get_workload(self):
return 'functional-query'
@classmethod
def add_test_dimensions(cls):
super(TestInsertWideTable, cls).add_test_dimensions()
# Only vary codegen
cls.ImpalaTestMatrix.add_dimension(create_exec_option_dimension(
cluster_sizes=[0], disable_codegen_options=[True, False], batch_sizes=[0]))
# Inserts only supported on text and parquet
# TODO: Enable 'text'/codec once the compressed text writers are in.
cls.ImpalaTestMatrix.add_constraint(lambda v:\
v.get_value('table_format').file_format == 'parquet' or \
v.get_value('table_format').file_format == 'text')
cls.ImpalaTestMatrix.add_constraint(lambda v:\
v.get_value('table_format').compression_codec == 'none')
# Don't run on core. This test is very slow (IMPALA-864) and we are unlikely to
# regress here.
if cls.exploration_strategy() == 'core':
cls.ImpalaTestMatrix.add_constraint(lambda v: False);
@SkipIfLocal.parquet_file_size
def test_insert_wide_table(self, vector, unique_database):
table_format = vector.get_value('table_format')
# Text can't handle as many columns as Parquet (codegen takes forever)
num_cols = 1000 if table_format.file_format == 'text' else 2000
table_name = unique_database + ".insert_widetable"
if vector.get_value('exec_option')['disable_codegen']:
table_name += "_codegen_disabled"
col_descs = widetable.get_columns(num_cols)
create_stmt = "CREATE TABLE " + table_name + "(" + ','.join(col_descs) + ")"
if vector.get_value('table_format').file_format == 'parquet':
create_stmt += " stored as parquet"
self.client.execute(create_stmt)
# Get a single row of data
col_vals = widetable.get_data(num_cols, 1, quote_strings=True)[0]
insert_stmt = "INSERT INTO " + table_name + " VALUES(" + col_vals + ")"
self.client.execute(insert_stmt)
result = self.client.execute("select count(*) from " + table_name)
assert result.data == ["1"]
result = self.client.execute("select * from " + table_name)
types = result.column_types
labels = result.column_labels
expected = QueryTestResult([col_vals], types, labels, order_matters=False)
actual = QueryTestResult(parse_result_rows(result), types, labels, order_matters=False)
assert expected == actual
class TestInsertPartKey(ImpalaTestSuite):
"""Regression test for IMPALA-875"""
@classmethod
def get_workload(self):
return 'functional-query'
@classmethod
def add_test_dimensions(cls):
super(TestInsertPartKey, cls).add_test_dimensions()
# Only run for a single table type
cls.ImpalaTestMatrix.add_dimension(create_exec_option_dimension(
cluster_sizes=[0], disable_codegen_options=[False], batch_sizes=[0],
sync_ddl=[1]))
cls.ImpalaTestMatrix.add_constraint(lambda v:
(v.get_value('table_format').file_format == 'text'))
cls.ImpalaTestMatrix.add_constraint(lambda v:\
v.get_value('table_format').compression_codec == 'none')
@pytest.mark.execute_serially
def test_insert_part_key(self, vector):
"""Test that partition column exprs are cast to the correct type. See IMPALA-875."""
self.run_test_case('QueryTest/insert_part_key', vector,
multiple_impalad=vector.get_value('exec_option')['sync_ddl'] == 1)
class TestInsertNullQueries(ImpalaTestSuite):
@classmethod
def get_workload(self):
return 'functional-query'
@classmethod
def add_test_dimensions(cls):
super(TestInsertNullQueries, cls).add_test_dimensions()
# Fix the exec_option vector to have a single value. This is needed should we decide
# to run the insert tests in parallel (otherwise there will be two tests inserting
# into the same table at the same time for the same file format).
# TODO: When we do decide to run these tests in parallel we could create unique temp
# tables for each test case to resolve the concurrency problems.
cls.ImpalaTestMatrix.add_dimension(create_exec_option_dimension(
cluster_sizes=[0], disable_codegen_options=[False], batch_sizes=[0]))
# These tests only make sense for inserting into a text table with special
# logic to handle all the possible ways NULL needs to be written as ascii
cls.ImpalaTestMatrix.add_constraint(lambda v:\
(v.get_value('table_format').file_format == 'text' and \
v.get_value('table_format').compression_codec == 'none'))
@classmethod
def setup_class(cls):
super(TestInsertNullQueries, cls).setup_class()
def test_insert_null(self, vector, unique_database):
self.run_test_case('QueryTest/insert_null', vector, unique_database,
test_file_vars={'$ORIGINAL_DB': ImpalaTestSuite
.get_db_name_from_format(vector.get_value('table_format'))})
class TestInsertFileExtension(ImpalaTestSuite):
"""Tests that files written to a table have the correct file extension. Asserts that
Parquet files end with .parq and text files end with .txt."""
@classmethod
def get_workload(self):
return 'functional-query'
@classmethod
def add_test_dimensions(cls):
super(TestInsertFileExtension, cls).add_test_dimensions()
cls.ImpalaTestMatrix.add_constraint(lambda v:
is_supported_insert_format(v.get_value('table_format')))
cls.ImpalaTestMatrix.add_dimension(create_single_exec_option_dimension())
@classmethod
def setup_class(cls):
super(TestInsertFileExtension, cls).setup_class()
def test_file_extension(self, vector, unique_database):
table_format = vector.get_value('table_format').file_format
if table_format == 'parquet':
file_extension = '.parq'
stored_as_format = 'parquet'
else:
file_extension = '.txt'
stored_as_format = 'textfile'
table_name = "{0}_table".format(table_format)
ctas_query = "create table {0}.{1} stored as {2} as select 1".format(
unique_database, table_name, stored_as_format)
self.execute_query_expect_success(self.client, ctas_query)
for path in self.filesystem_client.ls("test-warehouse/{0}.db/{1}".format(
unique_database, table_name)):
if not path.startswith('_'): assert path.endswith(file_extension)
class TestInsertHdfsWriterLimit(ImpalaTestSuite):
"""Test to make sure writer fragment instances are distributed evenly when using max
hdfs_writers query option."""
@classmethod
def get_workload(self):
return 'functional-query'
@classmethod
def add_test_dimensions(cls):
super(TestInsertHdfsWriterLimit, cls).add_test_dimensions()
cls.ImpalaTestMatrix.add_constraint(lambda v:
(v.get_value('table_format').file_format == 'parquet'))
@UniqueDatabase.parametrize(sync_ddl=True)
@SkipIfNotHdfsMinicluster.tuned_for_minicluster
def test_insert_writer_limit(self, unique_database):
# Root internal (non-leaf) fragment.
query = "create table {0}.test1 as select int_col from " \
"functional_parquet.alltypes".format(unique_database)
self.__run_insert_and_verify_instances(query, max_fs_writers=2, mt_dop=0,
expected_num_instances_per_host=[1, 2, 2])
# Root coordinator fragment.
query = "create table {0}.test2 as select int_col from " \
"functional_parquet.alltypes limit 100000".format(unique_database)
self.__run_insert_and_verify_instances(query, max_fs_writers=2, mt_dop=0,
expected_num_instances_per_host=[1, 1, 2])
# Root scan fragment. Instance count within limit.
query = "create table {0}.test3 as select int_col from " \
"functional_parquet.alltypes".format(unique_database)
self.__run_insert_and_verify_instances(query, max_fs_writers=4, mt_dop=0,
expected_num_instances_per_host=[1, 1, 1])
@UniqueDatabase.parametrize(sync_ddl=True)
@SkipIfNotHdfsMinicluster.tuned_for_minicluster
def test_mt_dop_writer_limit(self, unique_database):
# Root internal (non-leaf) fragment.
query = "create table {0}.test1 as select int_col from " \
"functional_parquet.alltypes".format(unique_database)
self.__run_insert_and_verify_instances(query, max_fs_writers=11, mt_dop=10,
expected_num_instances_per_host=[11, 12, 12])
# Root coordinator fragment.
query = "create table {0}.test2 as select int_col from " \
"functional_parquet.alltypes limit 100000".format(unique_database)
self.__run_insert_and_verify_instances(query, max_fs_writers=2, mt_dop=10,
expected_num_instances_per_host=[8, 8, 9])
# Root scan fragment. Instance count within limit.
query = "create table {0}.test3 as select int_col from " \
"functional_parquet.alltypes".format(unique_database)
self.__run_insert_and_verify_instances(query, max_fs_writers=30, mt_dop=10,
expected_num_instances_per_host=[8, 8, 8])
@UniqueDatabase.parametrize(sync_ddl=True)
@SkipIfNotHdfsMinicluster.tuned_for_minicluster
def test_processing_cost_writer_limit(self, unique_database):
"""Test both scenario of partitioned and unpartitioned insert.
All of the unpartitioned testscases will result in one instance writer because the
output volume is less than 256MB. Partitoned insert will result in 1 writer per node
unless max_fs_writers is set lower than num nodes."""
# Root internal (non-leaf) fragment.
query = "create table {0}.test1 as select int_col from " \
"functional_parquet.alltypes".format(unique_database)
self.__run_insert_and_verify_instances(query, max_fs_writers=2,
expected_num_instances_per_host=[1, 1, 2],
processing_cost_min_threads=1)
# Root coordinator fragment.
query = "create table {0}.test2 as select int_col from " \
"functional_parquet.alltypes limit 100000".format(unique_database)
self.__run_insert_and_verify_instances(query, max_fs_writers=2,
expected_num_instances_per_host=[1, 1, 2],
processing_cost_min_threads=1)
# Root internal (non-leaf) fragment. Instance count within limit.
query = "create table {0}.test3 as select int_col from " \
"functional_parquet.alltypes".format(unique_database)
self.__run_insert_and_verify_instances(query, max_fs_writers=30,
expected_num_instances_per_host=[1, 1, 2],
processing_cost_min_threads=1)
# Root internal (non-leaf) fragment. No max_fs_writers.
# Scan node and writer sink should always be in separate fragment with cost-based
# scaling.
query = "create table {0}.test4 as select int_col from " \
"functional_parquet.alltypes".format(unique_database)
self.__run_insert_and_verify_instances(query, max_fs_writers=0,
expected_num_instances_per_host=[1, 1, 2],
processing_cost_min_threads=1)
# Partitioned insert with 6 distinct partition values.
# Should create at least 1 writer per node.
query = "create table {0}.test5 partitioned by (ss_store_sk) as " \
"select ss_item_sk, ss_ticket_number, ss_store_sk " \
"from tpcds_parquet.store_sales".format(unique_database)
self.__run_insert_and_verify_instances(query, max_fs_writers=0,
expected_num_instances_per_host=[5, 5, 5],
processing_cost_min_threads=1)
# Partitioned insert can still be limited by max_fs_writers option.
query = "create table {0}.test6 partitioned by (ss_store_sk) as " \
"select ss_item_sk, ss_ticket_number, ss_store_sk " \
"from tpcds_parquet.store_sales".format(unique_database)
self.__run_insert_and_verify_instances(query, max_fs_writers=2,
expected_num_instances_per_host=[4, 5, 5],
processing_cost_min_threads=1)
def __run_insert_and_verify_instances(self, query, max_fs_writers=0, mt_dop=0,
processing_cost_min_threads=0,
expected_num_instances_per_host=[]):
self.client.set_configuration_option("max_fs_writers", max_fs_writers)
self.client.set_configuration_option("mt_dop", mt_dop)
if processing_cost_min_threads > 0:
self.client.set_configuration_option("compute_processing_cost", "true")
self.client.set_configuration_option("processing_cost_min_threads",
processing_cost_min_threads)
# Test depends on both planner and scheduler to see the same state of the cluster
# having 3 executors, so to reduce flakiness we make sure all 3 executors are up
# and running.
self.impalad_test_service.wait_for_metric_value("cluster-membership.backends.total",
3)
result = self.client.execute(query)
assert 'HDFS WRITER' in result.exec_summary[0]['operator'], result.runtime_profile
if (max_fs_writers > 0):
num_writers = int(result.exec_summary[0]['num_instances'])
assert (num_writers <= max_fs_writers), result.runtime_profile
regex = r'Per Host Number of Fragment Instances' \
r':.*?\((.*?)\).*?\((.*?)\).*?\((.*?)\).*?\n'
matches = re.findall(regex, result.runtime_profile)
assert len(matches) == 1 and len(matches[0]) == 3, result.runtime_profile
num_instances_per_host = [int(i) for i in matches[0]]
num_instances_per_host.sort()
expected_num_instances_per_host.sort()
assert num_instances_per_host == expected_num_instances_per_host, \
result.runtime_profile
self.client.clear_configuration()
class TestInsertNonPartitionedTable(ImpalaTestSuite):
@classmethod
def get_workload(self):
return 'functional-query'
@classmethod
def add_test_dimensions(cls):
super(TestInsertNonPartitionedTable, cls).add_test_dimensions()
cls.ImpalaTestMatrix.add_constraint(lambda v:
v.get_value('table_format').file_format == 'text'
and v.get_value('table_format').compression_codec == 'none')
cls.ImpalaTestMatrix.add_dimension(create_single_exec_option_dimension())
@classmethod
def setup_class(cls):
super(TestInsertNonPartitionedTable, cls).setup_class()
def test_insert_load_file_fail(self, vector, unique_database):
"""Tests metadata won't be corrupted after file metadata loading fails
in non-partitioned tables."""
table_name = '{0}.{1}'.format(unique_database, 'test_unpartition_tbl')
self.client.execute('create table {0}(f0 int)'
.format(table_name))
self.client.execute('insert overwrite table {0} select 0'
.format(table_name))
result = self.client.execute("select f0 from {0}".format(table_name))
assert result.data == ["0"]
exec_options = vector.get_value('exec_option')
exec_options['debug_action'] = 'catalogd_load_file_metadata_throw_exception'
try:
self.execute_query("insert overwrite table {0} select 1"
.format(table_name), exec_options)
assert False, "Expected query to fail."
except Exception as e:
assert "Failed to load metadata for table:" in str(e)
exec_options['debug_action'] = ''
self.execute_query("insert overwrite table {0} select 2"
.format(table_name), exec_options)
result = self.client.execute("select f0 from {0}".format(table_name))
assert result.data == ["2"]
class TestUnsafeImplicitCasts(ImpalaTestSuite):
"""Test to check 'allow_unsafe_casts' query-option behaviour on insert statements."""
@classmethod
def get_workload(cls):
return 'functional-query'
@classmethod
def add_test_dimensions(cls):
super(TestUnsafeImplicitCasts, cls).add_test_dimensions()
cls.ImpalaTestMatrix.add_constraint(lambda v:
(v.get_value('table_format').file_format == 'parquet'))
def test_unsafe_insert(self, vector, unique_database):
create_stmt = """create table {0}.unsafe_insert(tinyint_col tinyint,
smallint_col smallint, int_col int, bigint_col bigint, float_col float,
double_col double, decimal_col decimal, timestamp_col timestamp, date_col date,
string_col string, varchar_col varchar(100), char_col char(100),
bool_col boolean, binary_col binary)""".format(unique_database)
self.client.execute(create_stmt)
vector.get_value('exec_option')['allow_unsafe_casts'] = "true"
self.run_test_case('QueryTest/insert-unsafe', vector, unique_database)
Reference in New Issue View Git Blame Copy Permalink