You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by bo...@apache.org on 2022/05/26 13:21:10 UTC
[impala] branch master updated (0ee5f8084 -> 4236c307b)
This is an automated email from the ASF dual-hosted git repository.
boroknagyz pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
from 0ee5f8084 IMPALA-11317/IMPALA-11316/IMPALA-11315: impala-shell Python 3 fixes
new f77f57707 IMPALA-11251: Fix TestImpalaShellInteractive.test_unicode_input on Ubuntu 20
new 6a199be85 IMPALA-11249: Fix add_test_dimensions() locations to call super()
new 4236c307b IMPALA-10465: Use IGNORE variant of Kudu write operations
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
be/src/exec/kudu-table-sink.cc | 65 ++++-
be/src/exec/kudu-table-sink.h | 10 +
.../org/apache/impala/planner/KuduTableSink.java | 21 ++
tests/authorization/test_ranger.py | 8 +
tests/custom_cluster/test_client_ssl.py | 1 +
tests/custom_cluster/test_kudu.py | 320 ++++++++++++++++++---
tests/query_test/test_insert.py | 22 +-
tests/query_test/test_nested_types.py | 5 +-
tests/shell/test_shell_client.py | 11 +-
tests/shell/test_shell_commandline.py | 10 +-
tests/shell/test_shell_interactive.py | 18 +-
11 files changed, 439 insertions(+), 52 deletions(-)
[impala] 02/03: IMPALA-11249: Fix add_test_dimensions() locations to call super()
Posted by bo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
boroknagyz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 6a199be8544f5d76e00829f7493fcaac2f9c274c
Author: Joe McDonnell <jo...@cloudera.com>
AuthorDate: Sat May 21 19:10:11 2022 -0700
IMPALA-11249: Fix add_test_dimensions() locations to call super()
The original issue is that the strict HS2 shell tests
are not running in precommit or nightly jobs, but they
do run in local developer environments. Investigating
this showed that the shell tests were running with a
weird set of test dimensions that includes
table_format_and_file_extension. That dimension is only
used in test_insert.py::TestInsertFileExtension.
What is happening is that the shell tests and other
locations are running add_test_dimensions() without
calling super(..., cls).add_test_dimensions(). The
behavior is unclear, but there is clearly cross-talk
between the different tests that do this.
This changes all add_test_dimensions() locations to
call super(..., cls).add_test_dimensions() if they
don't already. Each location has been tuned to run
the same set of tests as before (except the shell
tests which now run the strict HS2 tests).
As part of this, several shell tests need to be
skipped or fixed for strict HS2.
Testing:
- Ran core job
- Ran tests locally to verify the set of tests
didn't change.
Change-Id: Ib20fd479d3b91ed0ed89a0bc5623cd2a5a458614
Reviewed-on: http://gerrit.cloudera.org:8080/18557
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
tests/authorization/test_ranger.py | 8 ++++++++
tests/custom_cluster/test_client_ssl.py | 1 +
tests/query_test/test_insert.py | 22 +++++++++++++++-------
tests/query_test/test_nested_types.py | 5 ++++-
tests/shell/test_shell_client.py | 11 +++++++++--
tests/shell/test_shell_commandline.py | 10 ++++++++--
tests/shell/test_shell_interactive.py | 16 +++++++++++++---
7 files changed, 58 insertions(+), 15 deletions(-)
diff --git a/tests/authorization/test_ranger.py b/tests/authorization/test_ranger.py
index 875886b12..d0caa7b17 100644
--- a/tests/authorization/test_ranger.py
+++ b/tests/authorization/test_ranger.py
@@ -1591,6 +1591,7 @@ class TestRangerColumnMaskingComplexTypesInSelectList(CustomClusterTestSuite):
@classmethod
def add_test_dimensions(cls):
+ super(TestRangerColumnMaskingComplexTypesInSelectList, cls).add_test_dimensions()
cls.ImpalaTestMatrix.add_dimension(create_client_protocol_dimension())
cls.ImpalaTestMatrix.add_dimension(create_orc_dimension(cls.get_workload()))
cls.ImpalaTestMatrix.add_constraint(lambda v:
@@ -1598,6 +1599,13 @@ class TestRangerColumnMaskingComplexTypesInSelectList(CustomClusterTestSuite):
cls.ImpalaTestMatrix.add_dimension(create_exec_option_dimension(
disable_codegen_options=[True]))
+ @classmethod
+ def add_custom_cluster_constraints(cls):
+ # Do not call the super() implementation, because this class needs to relax
+ # the set of constraints. The usual constraints only run on uncompressed text.
+ # This disables that constraint to let us run against only ORC.
+ return
+
@pytest.mark.execute_serially
@CustomClusterTestSuite.with_args(
impalad_args=IMPALAD_ARGS, catalogd_args=CATALOGD_ARGS)
diff --git a/tests/custom_cluster/test_client_ssl.py b/tests/custom_cluster/test_client_ssl.py
index 66a3fff78..4add40501 100644
--- a/tests/custom_cluster/test_client_ssl.py
+++ b/tests/custom_cluster/test_client_ssl.py
@@ -140,6 +140,7 @@ class TestClientSsl(CustomClusterTestSuite):
@classmethod
def add_test_dimensions(cls):
+ super(TestClientSsl, cls).add_test_dimensions()
cls.ImpalaTestMatrix.add_dimension(create_client_protocol_dimension())
@pytest.mark.execute_serially
diff --git a/tests/query_test/test_insert.py b/tests/query_test/test_insert.py
index 401d9b960..30d75fdc3 100644
--- a/tests/query_test/test_insert.py
+++ b/tests/query_test/test_insert.py
@@ -29,7 +29,9 @@ from tests.common.skip import SkipIfABFS, SkipIfEC, SkipIfLocal, \
SkipIfHive2, SkipIfNotHdfsMinicluster, SkipIfS3, SkipIfDockerizedCluster
from tests.common.test_dimensions import (
create_exec_option_dimension,
- create_uncompressed_text_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)
@@ -335,20 +337,26 @@ class TestInsertFileExtension(ImpalaTestSuite):
@classmethod
def add_test_dimensions(cls):
- cls.ImpalaTestMatrix.add_dimension(ImpalaTestDimension(
- 'table_format_and_file_extension',
- *[('parquet', '.parq'), ('textfile', '.txt')]))
+ 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_and_file_extension')[0]
- file_extension = vector.get_value('table_format_and_file_extension')[1]
+ 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, table_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)):
diff --git a/tests/query_test/test_nested_types.py b/tests/query_test/test_nested_types.py
index 3ca64f885..ef64e1181 100644
--- a/tests/query_test/test_nested_types.py
+++ b/tests/query_test/test_nested_types.py
@@ -36,7 +36,8 @@ from tests.common.skip import (
SkipIfNotHdfsMinicluster
)
from tests.common.test_dimensions import (create_exec_option_dimension,
- create_exec_option_dimension_from_dict, create_client_protocol_dimension)
+ create_exec_option_dimension_from_dict, create_client_protocol_dimension,
+ create_orc_dimension)
from tests.common.test_vector import ImpalaTestDimension
from tests.util.filesystem_utils import WAREHOUSE, get_fs_path, IS_HDFS
@@ -185,9 +186,11 @@ class TestNestedTypesInSelectListWithBeeswax(ImpalaTestSuite):
@classmethod
def add_test_dimensions(cls):
+ super(TestNestedTypesInSelectListWithBeeswax, cls).add_test_dimensions()
cls.ImpalaTestMatrix.add_dimension(create_client_protocol_dimension())
cls.ImpalaTestMatrix.add_constraint(lambda v:
v.get_value('protocol') == 'beeswax')
+ cls.ImpalaTestMatrix.add_dimension(create_orc_dimension(cls.get_workload()))
cls.ImpalaTestMatrix.add_dimension(create_exec_option_dimension(
disable_codegen_options=[True]))
diff --git a/tests/shell/test_shell_client.py b/tests/shell/test_shell_client.py
index 5fa1d5caa..1388b527f 100644
--- a/tests/shell/test_shell_client.py
+++ b/tests/shell/test_shell_client.py
@@ -20,8 +20,9 @@
from shell.impala_client import ImpalaBeeswaxClient, ImpalaHS2Client
from tests.common.impala_test_suite import ImpalaTestSuite
-from tests.common.test_dimensions import create_client_protocol_dimension
-from tests.common.test_dimensions import create_client_protocol_no_strict_dimension
+from tests.common.test_dimensions import (
+ create_client_protocol_dimension, create_client_protocol_no_strict_dimension,
+ create_uncompressed_text_dimension, create_single_exec_option_dimension)
from util import get_impalad_host_port
@@ -34,6 +35,12 @@ class TestShellClient(ImpalaTestSuite):
@classmethod
def add_test_dimensions(cls):
+ super(TestShellClient, cls).add_test_dimensions()
+ # Limit to uncompressed text with default exec options
+ cls.ImpalaTestMatrix.add_dimension(
+ create_uncompressed_text_dimension(cls.get_workload()))
+ cls.ImpalaTestMatrix.add_dimension(create_single_exec_option_dimension())
+ # Run with beeswax and HS2
cls.ImpalaTestMatrix.add_dimension(create_client_protocol_dimension())
cls.ImpalaTestMatrix.add_dimension(create_client_protocol_no_strict_dimension())
diff --git a/tests/shell/test_shell_commandline.py b/tests/shell/test_shell_commandline.py
index 3d698fb8a..3c6454d5a 100644
--- a/tests/shell/test_shell_commandline.py
+++ b/tests/shell/test_shell_commandline.py
@@ -34,8 +34,9 @@ from tests.common.environ import ImpalaTestClusterProperties
from tests.common.impala_service import ImpaladService
from tests.common.impala_test_suite import ImpalaTestSuite, IMPALAD_HS2_HOST_PORT
from tests.common.skip import SkipIf
-from tests.common.test_dimensions import create_client_protocol_dimension
-from tests.common.test_dimensions import create_client_protocol_strict_dimension
+from tests.common.test_dimensions import (
+ create_client_protocol_dimension, create_client_protocol_strict_dimension,
+ create_uncompressed_text_dimension, create_single_exec_option_dimension)
from time import sleep, time
from util import (get_impalad_host_port, assert_var_substitution, run_impala_shell_cmd,
ImpalaShell, IMPALA_SHELL_EXECUTABLE, SHELL_IS_PYTHON_2,
@@ -134,6 +135,11 @@ class TestImpalaShell(ImpalaTestSuite):
@classmethod
def add_test_dimensions(cls):
+ super(TestImpalaShell, cls).add_test_dimensions()
+ # Limit to uncompressed text with default exec options
+ cls.ImpalaTestMatrix.add_dimension(
+ create_uncompressed_text_dimension(cls.get_workload()))
+ cls.ImpalaTestMatrix.add_dimension(create_single_exec_option_dimension())
# Run with both beeswax and HS2 to ensure that behaviour is the same.
cls.ImpalaTestMatrix.add_dimension(create_client_protocol_dimension())
cls.ImpalaTestMatrix.add_dimension(create_client_protocol_strict_dimension())
diff --git a/tests/shell/test_shell_interactive.py b/tests/shell/test_shell_interactive.py
index c1411f86b..cd2db996b 100755
--- a/tests/shell/test_shell_interactive.py
+++ b/tests/shell/test_shell_interactive.py
@@ -40,8 +40,9 @@ from tempfile import NamedTemporaryFile
from tests.common.impala_service import ImpaladService
from tests.common.impala_test_suite import ImpalaTestSuite
from tests.common.skip import SkipIfLocal
-from tests.common.test_dimensions import create_client_protocol_dimension
-from tests.common.test_dimensions import create_client_protocol_strict_dimension
+from tests.common.test_dimensions import (
+ create_client_protocol_dimension, create_client_protocol_strict_dimension,
+ create_uncompressed_text_dimension, create_single_exec_option_dimension)
from tests.shell.util import get_unused_port
from util import (assert_var_substitution, ImpalaShell, get_impalad_port, get_shell_cmd,
get_open_sessions_metric, IMPALA_SHELL_EXECUTABLE, spawn_shell)
@@ -165,6 +166,11 @@ class TestImpalaShellInteractive(ImpalaTestSuite):
@classmethod
def add_test_dimensions(cls):
+ super(TestImpalaShellInteractive, cls).add_test_dimensions()
+ # Limit to uncompressed text with default exec options
+ cls.ImpalaTestMatrix.add_dimension(
+ create_uncompressed_text_dimension(cls.get_workload()))
+ cls.ImpalaTestMatrix.add_dimension(create_single_exec_option_dimension())
# Run with both beeswax and HS2 to ensure that behaviour is the same.
cls.ImpalaTestMatrix.add_dimension(create_client_protocol_dimension())
cls.ImpalaTestMatrix.add_dimension(create_client_protocol_strict_dimension())
@@ -681,6 +687,8 @@ class TestImpalaShellInteractive(ImpalaTestSuite):
def test_commandline_flag_disable_live_progress(self, vector):
"""Test the command line flag disable_live_progress with live_progress."""
+ if vector.get_value('strict_hs2_protocol'):
+ pytest.skip("Live option not supported in strict hs2 mode.")
# By default, shell option live_progress is set to True in the interactive mode.
cmds = "set all;"
result = run_impala_shell_interactive(vector, cmds)
@@ -704,13 +712,15 @@ class TestImpalaShellInteractive(ImpalaTestSuite):
cmds = "set all;"
# override the default option through command line argument.
- args = ['--strict_h2_protocol']
+ args = ['--strict_hs2_protocol']
result = run_impala_shell_interactive(vector, cmds, shell_args=args)
assert "\tLIVE_PROGRESS: False" in result.stdout
assert "\tLIVE_SUMMARY: False" in result.stdout
def test_live_option_configuration(self, vector):
"""Test the optional configuration file with live_progress and live_summary."""
+ if vector.get_value('strict_hs2_protocol'):
+ pytest.skip("Live option not supported in strict hs2 mode.")
# Positive tests
# set live_summary and live_progress as True with config file
rcfile_path = os.path.join(QUERY_FILE_PATH, 'good_impalarc3')
[impala] 03/03: IMPALA-10465: Use IGNORE variant of Kudu write operations
Posted by bo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
boroknagyz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 4236c307b971881a3b1d85068db5b053a9c34cfa
Author: Riza Suminto <ri...@cloudera.com>
AuthorDate: Sun May 8 22:53:20 2022 -0700
IMPALA-10465: Use IGNORE variant of Kudu write operations
KUDU-1563 added support for INSERT_IGNORE, UPDATE_IGNORE, and
DELETE_IGNORE to handle cases where users want to ignore primary key
errors efficiently. Impala already does this today for its INSERT
behavior. However, it does so by ignoring the per-row errors from Kudu
client side. This requires a large error buffer (which may need to be
expanded in rare cases) to log all of the warning messages which users
often do not care about and causes significant RPC overhead.
This patch change the Kudu write operation by Impala to use
INSERT_IGNORE, UPDATE_IGNORE, and DELETE_IGNORE if Kudu cluster supports
it and backend flag "kudu_ignore_conflicts" is true.
We benchmark the change by doing insert and update query on modified
tpch.lineitem table where we introduce conflicts for around half of the
total rows being modified. The table below shows the performance
difference after the patch:
+----------------------+--------+-------------+------------+------------+----------------+
| Query | Avg(s) | Base Avg(s) | Delta(Avg) | StdDev(%) | Base StdDev(%) |
+----------------------+--------+-------------+------------+------------+----------------+
| KUDU-IGNORE-3-UPDATE | 30.06 | 30.52 | -1.53% | 0.18% | 0.58% |
| KUDU-IGNORE-2-INSERT | 48.91 | 71.09 | I -31.20% | 0.60% | 0.72% |
+----------------------+--------+-------------+------------+------------+----------------+
Testing:
- Pass core tests.
Change-Id: I8da7c41d61b0888378b390b8b643238433eb3b52
Reviewed-on: http://gerrit.cloudera.org:8080/18536
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/exec/kudu-table-sink.cc | 65 ++++-
be/src/exec/kudu-table-sink.h | 10 +
.../org/apache/impala/planner/KuduTableSink.java | 21 ++
tests/custom_cluster/test_kudu.py | 320 ++++++++++++++++++---
4 files changed, 380 insertions(+), 36 deletions(-)
diff --git a/be/src/exec/kudu-table-sink.cc b/be/src/exec/kudu-table-sink.cc
index d2a167d64..1ba1ba986 100644
--- a/be/src/exec/kudu-table-sink.cc
+++ b/be/src/exec/kudu-table-sink.cc
@@ -21,6 +21,7 @@
#include <boost/bind.hpp>
#include <kudu/client/client.h>
+#include <kudu/client/resource_metrics.h>
#include <kudu/client/write_op.h>
#include <thrift/protocol/TDebugProtocol.h>
@@ -53,6 +54,23 @@ DEFINE_int32(kudu_error_buffer_size, DEFAULT_KUDU_ERROR_BUFFER_SIZE,
"The size (bytes) of the Kudu client buffer for returning errors, with a min of 1KB."
"If the actual errors exceed this size the query will fail.");
+// IMPALA-10465: This change the behavior of KuduTableSink to use the IGNORE variant of
+// Kudu API rather than using the non-IGNORE and discarding the Kudu error messages
+// manually. Set 'false' to return to the old behavior.
+DEFINE_bool(kudu_ignore_conflicts, true,
+ "Control whether Impala should ignore Kudu conflict error on duplicate and absent "
+ "primary keys during write operations. If this flag is set to true and Kudu cluster "
+ "supports ignore operations, Impala will use {INSERT,UPDATE,DELETE}_IGNORE "
+ "operations of Kudu API. Otherwise, Impala will use regular {INSERT,UPDATE,DELETE} "
+ "operations and logs any occurrences of such conflict error. "
+ "See also kudu_ignore_conflicts_in_transaction flag.");
+
+DEFINE_bool(kudu_ignore_conflicts_in_transaction, false,
+ "Control whether Kudu transaction should ignore conflict error on duplicate and "
+ "absent primary keys during write operations. If set to true, the Kudu transaction "
+ "will not be aborted for hitting such a conflict error. This flag is only "
+ "considered if kudu_ignore_conflicts flag is true.");
+
DECLARE_int32(kudu_operation_timeout_ms);
using kudu::client::KuduColumnSchema;
@@ -64,6 +82,7 @@ using kudu::client::KuduTransaction;
using kudu::client::KuduInsert;
using kudu::client::KuduUpdate;
using kudu::client::KuduError;
+using kudu::client::ResourceMetrics;
namespace impala {
@@ -179,6 +198,10 @@ Status KuduTableSink::Open(RuntimeState* state) {
session_ = client_->NewSession();
}
+ ignore_conflicts_ =
+ FLAGS_kudu_ignore_conflicts && kudu_table_sink_.ignore_not_found_or_duplicate;
+ if (!ignore_conflicts_) profile()->AppendExecOption("Log Kudu Conflicts");
+
session_->SetTimeoutMillis(FLAGS_kudu_operation_timeout_ms);
// KuduSession Set* methods here and below return a status for API compatibility.
@@ -236,6 +259,20 @@ kudu::client::KuduWriteOperation* KuduTableSink::NewWriteOp() {
}
}
+kudu::client::KuduWriteOperation* KuduTableSink::NewWriteIgnoreOp() {
+ if (sink_action_ == TSinkAction::INSERT) {
+ return table_->NewInsertIgnore();
+ } else if (sink_action_ == TSinkAction::UPDATE) {
+ return table_->NewUpdateIgnore();
+ } else if (sink_action_ == TSinkAction::UPSERT) {
+ return table_->NewUpsert();
+ } else {
+ DCHECK(sink_action_ == TSinkAction::DELETE)
+ << "Sink type not supported: " << sink_action_;
+ return table_->NewDeleteIgnore();
+ }
+}
+
Status KuduTableSink::Send(RuntimeState* state, RowBatch* batch) {
SCOPED_TIMER(profile()->total_time_counter());
expr_results_pool_->Clear();
@@ -249,10 +286,14 @@ Status KuduTableSink::Send(RuntimeState* state, RowBatch* batch) {
// violations.
int num_null_violations = 0;
+ RETURN_IF_ERROR(
+ DebugAction(state->query_options(), "FIS_KUDU_TABLE_SINK_WRITE_BEGIN"));
+
// Since everything is set up just forward everything to the writer.
for (int i = 0; i < batch->num_rows(); ++i) {
TupleRow* current_row = batch->GetRow(i);
- unique_ptr<kudu::client::KuduWriteOperation> write(NewWriteOp());
+ unique_ptr<kudu::client::KuduWriteOperation> write(
+ ignore_conflicts_ ? NewWriteIgnoreOp() : NewWriteOp());
bool add_row = true;
for (int j = 0; j < output_expr_evals_.size(); ++j) {
@@ -317,6 +358,27 @@ Status KuduTableSink::Send(RuntimeState* state, RowBatch* batch) {
Status KuduTableSink::CheckForErrors(RuntimeState* state) {
RETURN_IF_ERROR(state->CheckQueryState());
+
+ if (ignore_conflicts_) {
+ const ResourceMetrics& metrics = session_->GetWriteOpMetrics();
+ int64_t ignored_errors = metrics.GetMetric("insert_ignore_errors")
+ + metrics.GetMetric("update_ignore_errors")
+ + metrics.GetMetric("delete_ignore_errors");
+ DCHECK(ignored_errors >= total_ignored_errors_);
+ if (ignored_errors > total_ignored_errors_) {
+ total_ignored_errors_ = ignored_errors;
+ COUNTER_SET(num_row_errors_, total_ignored_errors_);
+ if (is_transactional_ && !FLAGS_kudu_ignore_conflicts_in_transaction) {
+ // Return general status error to abort transaction.
+ return Status("Kudu reported write operation errors during transaction.");
+ }
+ }
+ }
+
+ // Regardless of ignore_conflicts_ value, we still need to check for pending errors
+ // existence for the following possible scenarios:
+ // - sink_action_ is UPSERT.
+ // - Received other Kudu error such as timeouts or column constraint violations.
if (session_->CountPendingErrors() == 0) return Status::OK();
vector<KuduError*> errors;
@@ -380,6 +442,7 @@ Status KuduTableSink::FlushFinal(RuntimeState* state) {
void KuduTableSink::Close(RuntimeState* state) {
if (closed_) return;
session_.reset();
+ total_ignored_errors_ = 0;
mem_tracker_->Release(client_tracked_bytes_);
txn_.reset();
client_.reset();
diff --git a/be/src/exec/kudu-table-sink.h b/be/src/exec/kudu-table-sink.h
index 780d4eb8f..7b77ead4c 100644
--- a/be/src/exec/kudu-table-sink.h
+++ b/be/src/exec/kudu-table-sink.h
@@ -84,6 +84,9 @@ class KuduTableSink : public DataSink {
/// Create a new write operation according to the sink type.
kudu::client::KuduWriteOperation* NewWriteOp();
+ /// Create a new write ignore operation according to the sink type.
+ kudu::client::KuduWriteOperation* NewWriteIgnoreOp();
+
/// Checks for any errors buffered in the Kudu session, and increments
/// appropriate counters for ignored errors.
//
@@ -138,6 +141,13 @@ class KuduTableSink : public DataSink {
/// True if it's in Kudu transaction. It's valid only after Open() succeeds.
bool is_transactional_ = false;
+
+ /// True if this sink should ignore duplicate and absent key conflicts during Kudu
+ /// write operations. It's valid only after Open() succeeds.
+ bool ignore_conflicts_ = false;
+
+ /// Number of ignored write error operations during the lifetime of 'session_'.
+ int64_t total_ignored_errors_ = 0;
};
} // namespace impala
diff --git a/fe/src/main/java/org/apache/impala/planner/KuduTableSink.java b/fe/src/main/java/org/apache/impala/planner/KuduTableSink.java
index 21733ec17..eba08b5ce 100644
--- a/fe/src/main/java/org/apache/impala/planner/KuduTableSink.java
+++ b/fe/src/main/java/org/apache/impala/planner/KuduTableSink.java
@@ -24,6 +24,7 @@ import java.util.List;
import org.apache.impala.analysis.DescriptorTable;
import org.apache.impala.analysis.Expr;
import org.apache.impala.catalog.FeTable;
+import org.apache.impala.catalog.FeKuduTable;
import org.apache.impala.service.BackendConfig;
import org.apache.impala.thrift.TDataSink;
import org.apache.impala.thrift.TDataSinkType;
@@ -32,7 +33,12 @@ import org.apache.impala.thrift.TKuduTableSink;
import org.apache.impala.thrift.TQueryOptions;
import org.apache.impala.thrift.TTableSink;
import org.apache.impala.thrift.TTableSinkType;
+import org.apache.impala.util.KuduUtil;
+import org.apache.kudu.client.KuduClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
/**
@@ -40,6 +46,7 @@ import com.google.common.collect.Lists;
* data from a plan fragment into an Kudu table using a Kudu client.
*/
public class KuduTableSink extends TableSink {
+ private final static Logger LOG = LoggerFactory.getLogger(KuduTableSink.class);
// Optional list of referenced Kudu table column indices. The position of a result
// expression i matches a column index into the Kudu schema at targetColdIdxs[i].
@@ -49,6 +56,9 @@ public class KuduTableSink extends TableSink {
// target table is Kudu table and transaction for Kudu is enabled.
private java.nio.ByteBuffer txnToken_;
+ // Indicate whether Kudu cluster supports IGNORE write operations or not.
+ private boolean supportsIgnoreOperations_ = false;
+
public KuduTableSink(FeTable targetTable, Op sinkOp, List<Integer> referencedColumns,
List<Expr> outputExprs, java.nio.ByteBuffer txnToken) {
super(targetTable, sinkOp, outputExprs);
@@ -56,6 +66,16 @@ public class KuduTableSink extends TableSink {
? Lists.newArrayList(referencedColumns) : null;
txnToken_ =
txnToken != null ? org.apache.thrift.TBaseHelper.copyBinary(txnToken) : null;
+
+ // Check if Kudu cluster supports IGNORE write operations.
+ Preconditions.checkState(targetTable instanceof FeKuduTable);
+ KuduClient client =
+ KuduUtil.getKuduClient(((FeKuduTable) targetTable).getKuduMasterHosts());
+ try {
+ supportsIgnoreOperations_ = client.supportsIgnoreOperations();
+ } catch (Exception e) {
+ LOG.error("Unable to check Kudu ignore operation support", e);
+ }
}
@Override
@@ -95,6 +115,7 @@ public class KuduTableSink extends TableSink {
TKuduTableSink tKuduSink = new TKuduTableSink();
tKuduSink.setReferenced_columns(targetColIdxs_);
if (txnToken_ != null) tKuduSink.setKudu_txn_token(txnToken_);
+ tKuduSink.setIgnore_not_found_or_duplicate(supportsIgnoreOperations_);
tTableSink.setKudu_table_sink(tKuduSink);
tsink.table_sink = tTableSink;
tsink.output_exprs = Expr.treesToThrift(outputExprs_);
diff --git a/tests/custom_cluster/test_kudu.py b/tests/custom_cluster/test_kudu.py
index b5959a442..eda2c0731 100644
--- a/tests/custom_cluster/test_kudu.py
+++ b/tests/custom_cluster/test_kudu.py
@@ -334,9 +334,9 @@ class TestKuduHMSIntegration(CustomKuduTest):
self.run_test_case('QueryTest/kudu_hms_alter', vector, use_db=unique_database)
-class TestKuduTransaction(CustomClusterTestSuite):
+class TestKuduTransactionBase(CustomClusterTestSuite):
"""
- This suite tests the Kudu transaction when inserting rows to kudu table.
+ This is a base class of other TestKuduTransaction classes.
"""
# query to create Kudu table.
@@ -366,9 +366,7 @@ class TestKuduTransaction(CustomClusterTestSuite):
def get_workload(cls):
return 'functional-query'
- @pytest.mark.execute_serially
- @SkipIfKudu.no_hybrid_clock
- def test_kudu_txn_succeed(self, cursor, unique_database):
+ def _test_kudu_txn_succeed(self, cursor, unique_database):
# Create Kudu table.
table_name = "%s.test_kudu_txn_succeed" % unique_database
self.execute_query(self._create_kudu_table_query.format(table_name))
@@ -393,9 +391,7 @@ class TestKuduTransaction(CustomClusterTestSuite):
cursor.execute(self._row_num_query.format(table_name))
assert cursor.fetchall() == [(103,)]
- @pytest.mark.execute_serially
- @SkipIfKudu.no_hybrid_clock
- def test_kudu_txn_not_implemented(self, cursor, unique_database):
+ def _test_kudu_txn_not_implemented(self, cursor, unique_database):
# Create Kudu table.
table_name = "%s.test_kudu_txn_succeed" % unique_database
self.execute_query(self._create_kudu_table_query.format(table_name))
@@ -430,9 +426,8 @@ class TestKuduTransaction(CustomClusterTestSuite):
cursor.execute(self._row_num_query.format(table_name))
assert cursor.fetchall() == [(3,)]
- @pytest.mark.execute_serially
- @SkipIfKudu.no_hybrid_clock
- def test_kudu_txn_abort_dup_key(self, cursor, unique_database):
+ def _test_kudu_txn_abort_dup_key(self, cursor, unique_database,
+ expect_fail_on_conflict, expected_error_msg):
# Create Kudu table.
table_name = "%s.test_kudu_txn_abort_dup_key" % unique_database
self.execute_query(self._create_kudu_table_query.format(table_name))
@@ -442,11 +437,11 @@ class TestKuduTransaction(CustomClusterTestSuite):
self.execute_query("set ENABLE_KUDU_TRANSACTION=true")
try:
self.execute_query(self._insert_dup_key_query.format(table_name))
- assert False, "query was expected to fail"
+ assert (not expect_fail_on_conflict), "query was expected to fail"
except ImpalaBeeswaxException as e:
- assert "Key already present in Kudu table" in str(e)
+ assert expected_error_msg in str(e)
cursor.execute(self._row_num_query.format(table_name))
- assert cursor.fetchall() == [(0,)]
+ assert cursor.fetchall() == [(0 if expect_fail_on_conflict else 2,)]
# Disable Kudu transactions and run the same query. Part of rows are inserted into
# Kudu table.
@@ -470,11 +465,11 @@ class TestKuduTransaction(CustomClusterTestSuite):
# Transaction should be aborted and no rows are inserted into Kudu table.
try:
self.execute_query(self._insert_select_query2.format(table_name, table_name2))
- assert False, "query was expected to fail"
+ assert (not expect_fail_on_conflict), "query was expected to fail"
except ImpalaBeeswaxException as e:
- assert "Key already present in Kudu table" in str(e)
+ assert expected_error_msg in str(e)
cursor.execute(self._row_num_query.format(table_name))
- assert cursor.fetchall() == [(0,)]
+ assert cursor.fetchall() == [(0 if expect_fail_on_conflict else 2,)]
# Disable Kudu transactions and run the same query. Part of rows are inserted into
# Kudu table.
@@ -483,9 +478,8 @@ class TestKuduTransaction(CustomClusterTestSuite):
cursor.execute(self._row_num_query.format(table_name))
assert cursor.fetchall() == [(2,)]
- @pytest.mark.execute_serially
- @SkipIfKudu.no_hybrid_clock
- def test_kudu_txn_ctas(self, cursor, unique_database):
+ def _test_kudu_txn_ctas(self, cursor, unique_database, expect_fail_on_conflict,
+ expected_error_msg):
# Enable Kudu transactions
self.execute_query("set ENABLE_KUDU_TRANSACTION=true")
@@ -514,11 +508,11 @@ class TestKuduTransaction(CustomClusterTestSuite):
table_name4 = "%s.test_kudu_txn_ctas4" % unique_database
try:
self.execute_query(self._ctas_query.format(table_name4, table_name3))
- assert False, "query was expected to fail"
+ assert (not expect_fail_on_conflict), "query was expected to fail"
except ImpalaBeeswaxException as e:
- assert "Key already present in Kudu table" in str(e)
+ assert expected_error_msg in str(e)
cursor.execute(self._row_num_query.format(table_name4))
- assert cursor.fetchall() == [(0,)]
+ assert cursor.fetchall() == [(0 if expect_fail_on_conflict else 2,)]
# Disable Kudu transactions and run the same CTAS query. Part of rows are inserted
# into Kudu table.
@@ -528,10 +522,7 @@ class TestKuduTransaction(CustomClusterTestSuite):
cursor.execute(self._row_num_query.format(table_name5))
assert cursor.fetchall() == [(2,)]
- @pytest.mark.execute_serially
- @SkipIfKudu.no_hybrid_clock
- @SkipIfBuildType.not_dev_build
- def test_kudu_txn_abort_row_batch(self, cursor, unique_database):
+ def _test_kudu_txn_abort_row_batch(self, cursor, unique_database):
# Create Kudu table.
table_name = "%s.test_kudu_txn_abort_row_batch" % unique_database
self.execute_query(self._create_kudu_table_query.format(table_name))
@@ -549,10 +540,7 @@ class TestKuduTransaction(CustomClusterTestSuite):
cursor.execute(self._row_num_query.format(table_name))
assert cursor.fetchall() == [(0,)]
- @pytest.mark.execute_serially
- @SkipIfKudu.no_hybrid_clock
- @SkipIfBuildType.not_dev_build
- def test_kudu_txn_abort_partial_rows(self, cursor, unique_database):
+ def _test_kudu_txn_abort_partial_rows(self, cursor, unique_database):
# Create Kudu table.
table_name = "%s.test_kudu_txn_abort_partial_rows" % unique_database
self.execute_query(self._create_kudu_table_query.format(table_name))
@@ -570,10 +558,7 @@ class TestKuduTransaction(CustomClusterTestSuite):
cursor.execute(self._row_num_query.format(table_name))
assert cursor.fetchall() == [(0,)]
- @pytest.mark.execute_serially
- @SkipIfKudu.no_hybrid_clock
- @SkipIfBuildType.not_dev_build
- def test_kudu_txn_abort_partition_lock(self, cursor, unique_database):
+ def _test_kudu_txn_abort_partition_lock(self, cursor, unique_database):
# Running two separate queries that are inserting to the same Kudu partitions.
# Verify that one of the queries should fail, given Kudu's current implementation
# of partition locking.
@@ -603,6 +588,135 @@ class TestKuduTransaction(CustomClusterTestSuite):
self.client.close_query(handle)
+class TestKuduTransaction(TestKuduTransactionBase):
+ """
+ This suite tests the Kudu transaction when inserting rows to kudu table.
+ """
+
+ # expected error message from kudu on duplicate key.
+ _duplicate_key_error = "Kudu reported write operation errors during transaction."
+
+ @pytest.mark.execute_serially
+ @SkipIfKudu.no_hybrid_clock
+ def test_kudu_txn_succeed(self, cursor, unique_database):
+ self._test_kudu_txn_succeed(cursor, unique_database)
+
+ @pytest.mark.execute_serially
+ @SkipIfKudu.no_hybrid_clock
+ def test_kudu_txn_not_implemented(self, cursor, unique_database):
+ self._test_kudu_txn_not_implemented(cursor, unique_database)
+
+ @pytest.mark.execute_serially
+ @SkipIfKudu.no_hybrid_clock
+ def test_kudu_txn_abort_dup_key(self, cursor, unique_database):
+ self._test_kudu_txn_abort_dup_key(cursor, unique_database, True,
+ self._duplicate_key_error)
+
+ @pytest.mark.execute_serially
+ @SkipIfKudu.no_hybrid_clock
+ def test_kudu_txn_ctas(self, cursor, unique_database):
+ self._test_kudu_txn_ctas(cursor, unique_database, True, self._duplicate_key_error)
+
+ @pytest.mark.execute_serially
+ @SkipIfKudu.no_hybrid_clock
+ @SkipIfBuildType.not_dev_build
+ def test_kudu_txn_abort_row_batch(self, cursor, unique_database):
+ self._test_kudu_txn_abort_row_batch(cursor, unique_database)
+
+ @pytest.mark.execute_serially
+ @SkipIfKudu.no_hybrid_clock
+ @SkipIfBuildType.not_dev_build
+ def test_kudu_txn_abort_partial_rows(self, cursor, unique_database):
+ self._test_kudu_txn_abort_partial_rows(cursor, unique_database)
+
+ @pytest.mark.execute_serially
+ @SkipIfKudu.no_hybrid_clock
+ @SkipIfBuildType.not_dev_build
+ def test_kudu_txn_abort_partition_lock(self, cursor, unique_database):
+ self._test_kudu_txn_abort_partial_rows(cursor, unique_database)
+
+
+class TestKuduTransactionNoIgnore(TestKuduTransactionBase):
+ """
+ This suite tests the Kudu transaction when inserting rows to kudu table with
+ kudu_ignore_conflicts flag set to false.
+ """
+
+ # impalad args to start the cluster.
+ _impalad_args = "--kudu_ignore_conflicts=false"
+ # expected error message from kudu on duplicated key.
+ _duplicate_key_error = "Key already present in Kudu table"
+
+ @pytest.mark.execute_serially
+ @SkipIfKudu.no_hybrid_clock
+ @CustomClusterTestSuite.with_args(impalad_args=_impalad_args)
+ def test_kudu_txn_succeed(self, cursor, unique_database):
+ self._test_kudu_txn_succeed(cursor, unique_database)
+
+ @pytest.mark.execute_serially
+ @SkipIfKudu.no_hybrid_clock
+ @CustomClusterTestSuite.with_args(impalad_args=_impalad_args)
+ def test_kudu_txn_not_implemented(self, cursor, unique_database):
+ self._test_kudu_txn_not_implemented(cursor, unique_database)
+
+ @pytest.mark.execute_serially
+ @SkipIfKudu.no_hybrid_clock
+ @CustomClusterTestSuite.with_args(impalad_args=_impalad_args)
+ def test_kudu_txn_abort_dup_key(self, cursor, unique_database):
+ self._test_kudu_txn_abort_dup_key(cursor, unique_database, True,
+ self._duplicate_key_error)
+
+ @pytest.mark.execute_serially
+ @SkipIfKudu.no_hybrid_clock
+ @CustomClusterTestSuite.with_args(impalad_args=_impalad_args)
+ def test_kudu_txn_ctas(self, cursor, unique_database):
+ self._test_kudu_txn_ctas(cursor, unique_database, True, self._duplicate_key_error)
+
+ @pytest.mark.execute_serially
+ @SkipIfKudu.no_hybrid_clock
+ @SkipIfBuildType.not_dev_build
+ @CustomClusterTestSuite.with_args(impalad_args=_impalad_args)
+ def test_kudu_txn_abort_row_batch(self, cursor, unique_database):
+ self._test_kudu_txn_abort_row_batch(cursor, unique_database)
+
+ @pytest.mark.execute_serially
+ @SkipIfKudu.no_hybrid_clock
+ @SkipIfBuildType.not_dev_build
+ @CustomClusterTestSuite.with_args(impalad_args=_impalad_args)
+ def test_kudu_txn_abort_partial_rows(self, cursor, unique_database):
+ self._test_kudu_txn_abort_partial_rows(cursor, unique_database)
+
+ @pytest.mark.execute_serially
+ @SkipIfKudu.no_hybrid_clock
+ @SkipIfBuildType.not_dev_build
+ @CustomClusterTestSuite.with_args(impalad_args=_impalad_args)
+ def test_kudu_txn_abort_partition_lock(self, cursor, unique_database):
+ self._test_kudu_txn_abort_partial_rows(cursor, unique_database)
+
+
+class TestKuduTransactionIgnoreConflict(TestKuduTransactionBase):
+ """
+ This suite tests the Kudu transaction when inserting rows to kudu table with
+ kudu_ignore_conflicts=true and kudu_ignore_conflicts_in_transaction=true.
+ """
+
+ # impalad args to start the cluster.
+ _impalad_args = "--kudu_ignore_conflicts=true " \
+ "--kudu_ignore_conflicts_in_transaction=true"
+
+ @pytest.mark.execute_serially
+ @SkipIfKudu.no_hybrid_clock
+ @CustomClusterTestSuite.with_args(impalad_args=_impalad_args)
+ def test_kudu_txn_dup_key(self, cursor, unique_database):
+ self._test_kudu_txn_abort_dup_key(cursor, unique_database, False, "no error")
+
+ @pytest.mark.execute_serially
+ @SkipIfKudu.no_hybrid_clock
+ @CustomClusterTestSuite.with_args(impalad_args=_impalad_args)
+ def test_kudu_txn_ctas(self, cursor, unique_database):
+ self._test_kudu_txn_ctas(cursor, unique_database, False, "no error")
+
+
class TestKuduTxnKeepalive(CustomKuduTest):
"""
Tests the Kudu transaction to ensure the transaction handle kept by the front-end in
@@ -649,3 +763,139 @@ class TestKuduTxnKeepalive(CustomKuduTest):
self.execute_query(self._insert_3_rows_query.format(table_name), query_options)
cursor.execute(self._row_num_query.format(table_name))
assert cursor.fetchall() == [(3,)]
+
+
+class TestKuduDmlConflictBase(CustomClusterTestSuite):
+ """
+ This is a base class of other TestKuduDml classes.
+ """
+
+ # query to create Kudu table.
+ _create_kudu_table_query = ("create table {0} "
+ "(a int primary key, b timestamp not null) "
+ "partition by hash(a) partitions 8 stored as kudu")
+ # queries to insert rows into Kudu table.
+ _insert_dup_key_query = ("insert into {0} values "
+ "(0, '1400-01-01'), (0, '1400-01-02'), "
+ "(1, '1400-01-01'), (1, '1400-01-02'), "
+ "(2, '1400-01-01'), (2, '1400-01-02')")
+ # query to update rows in Kudu table with some constraint violation.
+ _update_violate_constraint_query = ("update {0} set b = "
+ "case when b <= '2022-01-01' then NULL else '1400-01-01' end")
+ # query to update row by primary key.
+ _update_by_key_query = "update {0} set b = '1400-02-02' where a = {1}"
+ # query to delete row by primary key.
+ _delete_by_key_query = "delete from {0} where a = {1}"
+ # query to drop all rows from Kudu table.
+ _delete_all_query = "delete from {0}"
+ # query to get number of rows.
+ _row_num_query = "select count(*) from {0}"
+
+ @classmethod
+ def get_workload(cls):
+ return 'functional-query'
+
+ def _check_errors(self, query_profile, expect_error, error_message, num_row_erros):
+ """
+ Check ocurrence of error_message and num_row_errors in query_profile.
+ """
+
+ error_line = " Errors: {0}".format(error_message)
+ num_row_error_line = "NumRowErrors: {0}".format(num_row_erros)
+ assert expect_error == (error_line in query_profile)
+ assert num_row_error_line in query_profile
+
+ def _race_queries(self, fast_query, slow_query, expect_error_on_slow_query,
+ error_message, num_row_erros):
+ """
+ Race two queries and check for error message in the slow query's profile.
+ """
+
+ fast_sleep = {'debug_action': 'FIS_KUDU_TABLE_SINK_WRITE_BEGIN:SLEEP@1000'}
+ slow_sleep = {'debug_action': 'FIS_KUDU_TABLE_SINK_WRITE_BEGIN:SLEEP@3000'}
+ timeout = 10
+
+ fast_handle = self.execute_query_async(fast_query, fast_sleep)
+ slow_handle = self.execute_query_async(slow_query, slow_sleep)
+ try:
+ # Wait for both queries to finish.
+ self.wait_for_state(fast_handle, self.client.QUERY_STATES['FINISHED'], timeout)
+ self.wait_for_state(slow_handle, self.client.QUERY_STATES['FINISHED'], timeout)
+ self._check_errors(self.client.get_runtime_profile(slow_handle),
+ expect_error_on_slow_query, error_message, num_row_erros)
+ finally:
+ self.client.close_query(fast_handle)
+ self.client.close_query(slow_handle)
+
+ def _test_insert_update_delete(self, cursor, unique_database,
+ expect_log_on_conflict):
+ """
+ Do sequence of insert, update, and delete query with conflicting primary keys.
+ """
+
+ # Create Kudu table.
+ table_name = "%s.insert_update_delete" % unique_database
+ self.execute_query(self._create_kudu_table_query.format(table_name))
+
+ # Insert rows with duplicate primary key.
+ # Error message should exist in profile if kudu_ignore_conflicts=true.
+ result = self.execute_query(self._insert_dup_key_query.format(table_name))
+ self._check_errors(result.runtime_profile, expect_log_on_conflict,
+ "Key already present in Kudu table", 3)
+ cursor.execute(self._row_num_query.format(table_name))
+ assert cursor.fetchall() == [(3,)]
+
+ # Update rows with some constraint violation.
+ # Error message should exist in profile regardless of kudu_ignore_conflicts value.
+ result = self.execute_query(self._update_violate_constraint_query.format(table_name))
+ self._check_errors(result.runtime_profile, True,
+ "Row with null value violates nullability constraint on table", 3)
+
+ # Update row with non-existent primary key by racing it against concurrent delete
+ # query.
+ delete_query = self._delete_by_key_query.format(table_name, 1)
+ update_query = self._update_by_key_query.format(table_name, 1)
+ self._race_queries(delete_query, update_query, expect_log_on_conflict,
+ "Not found in Kudu table", 1)
+ cursor.execute(self._row_num_query.format(table_name))
+ assert cursor.fetchall() == [(2,)]
+
+ # Delete row with non-existent primary key by racing it against another concurrent
+ # delete.
+ delete_query = self._delete_by_key_query.format(table_name, 2)
+ self._race_queries(delete_query, delete_query, expect_log_on_conflict,
+ "Not found in Kudu table", 1)
+ cursor.execute(self._row_num_query.format(table_name))
+ assert cursor.fetchall() == [(1,)]
+
+ # Delete all rows. Expect no errors.
+ result = self.execute_query(self._delete_all_query.format(table_name))
+ self._check_errors(result.runtime_profile, True, "\n", 0)
+ cursor.execute(self._row_num_query.format(table_name))
+ assert cursor.fetchall() == [(0,)]
+
+
+class TestKuduDmlConflictNoError(TestKuduDmlConflictBase):
+ """
+ Test that Kudu DML ignore conflict and does not log the conflict error message.
+ """
+
+ @pytest.mark.execute_serially
+ @SkipIfKudu.no_hybrid_clock
+ def test_insert_update_delete(self, cursor, unique_database):
+ self._test_insert_update_delete(cursor, unique_database, False)
+
+
+class TestKuduDmlConflictLogError(TestKuduDmlConflictBase):
+ """
+ Test that Kudu DML not ignore conflict and log the conflict error message.
+ """
+
+ # impalad args to start the cluster.
+ _impalad_args = "--kudu_ignore_conflicts=false"
+
+ @pytest.mark.execute_serially
+ @SkipIfKudu.no_hybrid_clock
+ @CustomClusterTestSuite.with_args(impalad_args=_impalad_args)
+ def test_insert_update_delete(self, cursor, unique_database):
+ self._test_insert_update_delete(cursor, unique_database, True)
[impala] 01/03: IMPALA-11251: Fix TestImpalaShellInteractive.test_unicode_input on Ubuntu 20
Posted by bo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
boroknagyz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit f77f57707417379635880efa1caa1ec40f2314b2
Author: Joe McDonnell <jo...@cloudera.com>
AuthorDate: Sat May 21 19:28:10 2022 -0700
IMPALA-11251: Fix TestImpalaShellInteractive.test_unicode_input on Ubuntu 20
TestImpalaShellInteractive.test_unicode_input() was spawning
impala-shell using a command that preserves the environment. This
is a problem on Ubuntu 20, because Ubuntu 20 uses a newer GCC/libstdc++.
Preserving the environment keeps an LD_LIBRARY_PATH setting that
uses an older libstdc++ and causes the shell to fail at startup.
This switches the test to use shell/util.py's spawn_shell(),
which cleans up the environment before running the shell.
Testing:
- Ran tests on Ubuntu 20
Change-Id: Ib07f557ab3c21d6b39f814dcfc0bf9eb1b61f090
Reviewed-on: http://gerrit.cloudera.org:8080/18558
Reviewed-by: Joe McDonnell <jo...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
tests/shell/test_shell_interactive.py | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/tests/shell/test_shell_interactive.py b/tests/shell/test_shell_interactive.py
index 2d1a4aa7a..c1411f86b 100755
--- a/tests/shell/test_shell_interactive.py
+++ b/tests/shell/test_shell_interactive.py
@@ -373,7 +373,7 @@ class TestImpalaShellInteractive(ImpalaTestSuite):
# IMPALA-10415: Multiline query with history enabled and unicode chars.
# readline gets its input from tty, so using stdin does not work.
shell_cmd = get_shell_cmd(vector)
- child_proc = pexpect.spawn(shell_cmd[0], shell_cmd[1:])
+ child_proc = spawn_shell(shell_cmd)
child_proc.expect(PROMPT_REGEX)
child_proc.sendline("select '{0}'\n;".format(unicode_bytes))
child_proc.expect("Fetched 1 row\(s\) in [0-9]+\.?[0-9]*s")