You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by ja...@apache.org on 2025/08/19 14:35:49 UTC
(impala) branch master updated (0cd240a1c -> 75ca356ed)
This is an automated email from the ASF dual-hosted git repository.
jasonmfehr pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
from 0cd240a1c IMPALA-14312: Fix Issues with ssl_minimum_version Flag Validator
new c23d3cb58 IMPALA-14311: Fix ASAN issue in test_query_cancel_load_tables
new 80f5e3bfa IMPALA-14322: Fix typo in IMPALA-12520
new 75ca356ed IMPALA-13237: [Patch 6] - Match Behavior of Older Impala Versions
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/observe/span-manager.cc | 7 +-
.../queries/QueryTest/iceberg-alter-default.test | 2 +-
.../queries/QueryTest/iceberg-alter-v1.test | 2 +-
.../queries/QueryTest/iceberg-alter-v2.test | 2 +-
.../queries/QueryTest/iceberg-create.test | 14 +--
.../queries/QueryTest/iceberg-negative.test | 2 +-
tests/custom_cluster/test_otel_trace.py | 106 ++++++++++-----------
tests/custom_cluster/test_query_expiration.py | 10 +-
tests/util/query_profile_util.py | 7 ++
tests/util/web_pages_util.py | 18 ++--
tests/webserver/test_web_pages.py | 6 +-
11 files changed, 92 insertions(+), 84 deletions(-)
(impala) 01/03: IMPALA-14311: Fix ASAN issue in test_query_cancel_load_tables
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jasonmfehr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit c23d3cb58adc558fe5e5064f9db6f6471c18d4ea
Author: Riza Suminto <ri...@cloudera.com>
AuthorDate: Fri Aug 15 15:30:31 2025 -0700
IMPALA-14311: Fix ASAN issue in test_query_cancel_load_tables
verifiers.test_verify_metrics.TestValidateMetrics.test_metrics_are_zero
fails in ASAN tests. This is because a client session opened by
test_query_cancel_load_tables is not cleanly closed.
This patch fix the issue by creating the Impala client within the run
method of web_pages_util.py using with-as statement.
This patch also attempt to fix flakiness in test_query_expiration by
validating that time_limit_expire_handle is actually closed in
Coordinator log (IMPALA-13444).
Testing:
- Pass test_web_pages.py test in ASAN build.
- Loop and pass test_query_expiration 50x in ASAN build.
Change-Id: I6fcf13902478535f6fa15f267edc83891057993c
Reviewed-on: http://gerrit.cloudera.org:8080/23302
Reviewed-by: Daniel Becker <da...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
tests/custom_cluster/test_query_expiration.py | 10 ++++++++--
tests/util/web_pages_util.py | 18 +++++++++++-------
tests/webserver/test_web_pages.py | 6 +++---
3 files changed, 22 insertions(+), 12 deletions(-)
diff --git a/tests/custom_cluster/test_query_expiration.py b/tests/custom_cluster/test_query_expiration.py
index 3f088713f..ceffee6a3 100644
--- a/tests/custom_cluster/test_query_expiration.py
+++ b/tests/custom_cluster/test_query_expiration.py
@@ -119,9 +119,15 @@ class TestQueryExpiration(CustomClusterTestSuite):
r"minimum memory reservation is greater than memory available.*\nQuery "
+ r"[0-9a-f]+:[0-9a-f]+ expired due to client inactivity \(timeout is 1s000ms\)")
# hs2 client does not automaticaly close time_limit_expire_handle.
- # manually close it.
+ # manually close it. It can take few miliseconds in slow build (ie., ASAN) until
+ # query actually closed in the coordinator. So make sure from the log that it is
+ # actually closed.
+ tle_query_id = client.get_query_id(time_limit_expire_handle)
client.close_query(time_limit_expire_handle)
- self._check_num_executing(impalad, 2)
+ self.assert_impalad_log_contains(
+ 'INFO', r'ReleaseQueryState\(\): deleted query_id=' + tle_query_id,
+ timeout_s=4)
+ self._check_num_executing(impalad, 2, 0)
# Both queries with query_timeout_s < 4 should generate this message.
self.assert_impalad_log_contains('INFO', "Expiring query due to client inactivity: "
r"[0-9a-f]+:[0-9a-f]+, last activity was at: \d\d\d\d-\d\d-\d\d \d\d:\d\d:\d\d",
diff --git a/tests/util/web_pages_util.py b/tests/util/web_pages_util.py
index 4b294d893..cb98d5d76 100644
--- a/tests/util/web_pages_util.py
+++ b/tests/util/web_pages_util.py
@@ -22,6 +22,7 @@ import requests
from tests.util.retry import retry
+
def get_num_completed_backends(service, query_id):
"""Get the number of completed backends for the given query_id from the
'query_backends' web page."""
@@ -92,18 +93,21 @@ def assert_query_stopped(impalad, query_id):
assert len(expected_queries) == 1
-def run(client, query, queue):
+def run(test_class, query, options, queue):
"""Execute query and put results or errors into queue."""
- try:
- queue.put(str(client.execute(query)))
- except Exception as ex:
- queue.put(str(ex))
+ with test_class.create_impala_client() as client:
+ if options:
+ client.set_configuration(options)
+ try:
+ queue.put(str(client.execute(query)))
+ except Exception as ex:
+ queue.put(str(ex))
-def start(client, query):
+def start(test_class, query, options=None):
"""Execute a query in a separate process. Returns (process, queue)."""
queue = Queue()
- proc = Process(target=run, args=(client, query, queue))
+ proc = Process(target=run, args=(test_class, query, options, queue))
proc.start()
return proc, queue
diff --git a/tests/webserver/test_web_pages.py b/tests/webserver/test_web_pages.py
index c80a77f20..4a8d5fa10 100644
--- a/tests/webserver/test_web_pages.py
+++ b/tests/webserver/test_web_pages.py
@@ -1087,7 +1087,7 @@ class TestWebPage(ImpalaTestSuite):
# Start the query completely async. The server doesn't return a response until
# the query has exited the CREATED state, so we need to get the query ID another way.
- proc, queue = start(self.client, "select count(*) from functional_parquet.alltypes")
+ proc, queue = start(self, "select count(*) from functional_parquet.alltypes")
wait_for_state(impalad, 'CREATED')
in_flight_queries = impalad.get_debug_webpage_json('queries')['in_flight_queries']
@@ -1113,11 +1113,11 @@ class TestWebPage(ImpalaTestSuite):
impalad = ImpalaCluster.get_e2e_test_cluster().impalads[0].service
wait_for_state(impalad, None)
delay_created_action = "impalad_load_tables_delay:SLEEP@5000"
- self.client.set_configuration({'debug_action': delay_created_action})
# Start the query completely async. The server doesn't return a response until
# the query has exited the CREATED state, so we need to get the query ID another way.
- proc, queue = start(self.client, "select count(*) from functional_parquet.alltypes")
+ proc, queue = start(self, "select count(*) from functional_parquet.alltypes",
+ {'debug_action': delay_created_action})
wait_for_state(impalad, 'CREATED')
in_flight_queries = impalad.get_debug_webpage_json('queries')['in_flight_queries']
(impala) 02/03: IMPALA-14322: Fix typo in IMPALA-12520
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jasonmfehr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 80f5e3bfa398490224fbd06a170b268fe1717a51
Author: Daniel Vanko <dv...@cloudera.com>
AuthorDate: Mon Aug 18 15:42:45 2025 +0200
IMPALA-14322: Fix typo in IMPALA-12520
In IMPALA-12520 affected paths were modified to test_warehouse
(separated with underscore) instead of test-warehouse (with hyphen).
This commit replaces the underscore to hyphen.
Change-Id: I3a9737af3e6169cc0cd144df53fd35e9e2b20468
Reviewed-on: http://gerrit.cloudera.org:8080/23304
Reviewed-by: Daniel Becker <da...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
.../queries/QueryTest/iceberg-alter-default.test | 2 +-
.../queries/QueryTest/iceberg-alter-v1.test | 2 +-
.../queries/QueryTest/iceberg-alter-v2.test | 2 +-
.../functional-query/queries/QueryTest/iceberg-create.test | 14 +++++++-------
.../queries/QueryTest/iceberg-negative.test | 2 +-
5 files changed, 11 insertions(+), 11 deletions(-)
diff --git a/testdata/workloads/functional-query/queries/QueryTest/iceberg-alter-default.test b/testdata/workloads/functional-query/queries/QueryTest/iceberg-alter-default.test
index 7b51dd6e4..3a16b4a35 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/iceberg-alter-default.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/iceberg-alter-default.test
@@ -74,7 +74,7 @@ CREATE TABLE iceberg_hadoop_catalog(
)
STORED AS ICEBERG
TBLPROPERTIES('iceberg.catalog'='hadoop.catalog',
-'iceberg.catalog_location'='$WAREHOUSE_LOCATION_PREFIX/test_warehouse/$DATABASE/hadoop_catalog_test');
+'iceberg.catalog_location'='$WAREHOUSE_LOCATION_PREFIX/test-warehouse/$DATABASE/hadoop_catalog_test');
ALTER TABLE iceberg_hadoop_catalog ADD COLUMNS(event_time TIMESTAMP, register_time DATE);
ALTER TABLE iceberg_hadoop_catalog ADD COLUMNS(message STRING, price DECIMAL(8,1));
ALTER TABLE iceberg_hadoop_catalog ADD COLUMNS(map_test MAP <STRING, array <STRING>>, struct_test STRUCT <f1: BIGINT, f2: BIGINT>);
diff --git a/testdata/workloads/functional-query/queries/QueryTest/iceberg-alter-v1.test b/testdata/workloads/functional-query/queries/QueryTest/iceberg-alter-v1.test
index 9aa74288a..1fe2c3525 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/iceberg-alter-v1.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/iceberg-alter-v1.test
@@ -74,7 +74,7 @@ CREATE TABLE iceberg_hadoop_catalog(
)
STORED AS ICEBERG
TBLPROPERTIES('format-version'='1', 'iceberg.catalog'='hadoop.catalog',
-'iceberg.catalog_location'='$WAREHOUSE_LOCATION_PREFIX/test_warehouse/$DATABASE/hadoop_catalog_test');
+'iceberg.catalog_location'='$WAREHOUSE_LOCATION_PREFIX/test-warehouse/$DATABASE/hadoop_catalog_test');
ALTER TABLE iceberg_hadoop_catalog ADD COLUMNS(event_time TIMESTAMP, register_time DATE);
ALTER TABLE iceberg_hadoop_catalog ADD COLUMNS(message STRING, price DECIMAL(8,1));
ALTER TABLE iceberg_hadoop_catalog ADD COLUMNS(map_test MAP <STRING, array <STRING>>, struct_test STRUCT <f1: BIGINT, f2: BIGINT>);
diff --git a/testdata/workloads/functional-query/queries/QueryTest/iceberg-alter-v2.test b/testdata/workloads/functional-query/queries/QueryTest/iceberg-alter-v2.test
index 5e512f970..024b1a9a7 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/iceberg-alter-v2.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/iceberg-alter-v2.test
@@ -74,7 +74,7 @@ CREATE TABLE iceberg_hadoop_catalog(
)
STORED AS ICEBERG
TBLPROPERTIES('format-version'='2', 'iceberg.catalog'='hadoop.catalog',
-'iceberg.catalog_location'='$WAREHOUSE_LOCATION_PREFIX/test_warehouse/$DATABASE/hadoop_catalog_test');
+'iceberg.catalog_location'='$WAREHOUSE_LOCATION_PREFIX/test-warehouse/$DATABASE/hadoop_catalog_test');
ALTER TABLE iceberg_hadoop_catalog ADD COLUMNS(event_time TIMESTAMP, register_time DATE);
ALTER TABLE iceberg_hadoop_catalog ADD COLUMNS(message STRING, price DECIMAL(8,1));
ALTER TABLE iceberg_hadoop_catalog ADD COLUMNS(map_test MAP <STRING, array <STRING>>, struct_test STRUCT <f1: BIGINT, f2: BIGINT>);
diff --git a/testdata/workloads/functional-query/queries/QueryTest/iceberg-create.test b/testdata/workloads/functional-query/queries/QueryTest/iceberg-create.test
index 451c81771..49b84ac7b 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/iceberg-create.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/iceberg-create.test
@@ -70,13 +70,13 @@ PARTITIONED BY SPEC
TRUNCATE(15, level)
)
STORED AS ICEBERG
-LOCATION '$WAREHOUSE_LOCATION_PREFIX/test_warehouse/$DATABASE.iceberg_test_with_location'
+LOCATION '$WAREHOUSE_LOCATION_PREFIX/test-warehouse/$DATABASE.iceberg_test_with_location'
TBLPROPERTIES('iceberg.catalog'='hadoop.tables');
CREATE EXTERNAL TABLE iceberg_hadoop_tbls_external(
level STRING
)
STORED AS ICEBERG
-LOCATION '$WAREHOUSE_LOCATION_PREFIX/test_warehouse/$DATABASE.iceberg_test_with_location'
+LOCATION '$WAREHOUSE_LOCATION_PREFIX/test-warehouse/$DATABASE.iceberg_test_with_location'
TBLPROPERTIES('iceberg.catalog'='hadoop.tables');
---- RESULTS
'Table has been created.'
@@ -97,7 +97,7 @@ STRING,BIGINT,BIGINT
---- QUERY
CREATE EXTERNAL TABLE iceberg_hadoop_tbls_external_empty_col
STORED AS ICEBERG
-LOCATION '$WAREHOUSE_LOCATION_PREFIX/test_warehouse/$DATABASE.iceberg_test_with_location'
+LOCATION '$WAREHOUSE_LOCATION_PREFIX/test-warehouse/$DATABASE.iceberg_test_with_location'
TBLPROPERTIES('iceberg.catalog'='hadoop.tables');
---- RESULTS
'Table has been created.'
@@ -137,7 +137,7 @@ PARTITIONED BY SPEC
)
STORED AS ICEBERG
TBLPROPERTIES('iceberg.catalog'='hadoop.catalog',
-'iceberg.catalog_location'='$WAREHOUSE_LOCATION_PREFIX/test_warehouse/$DATABASE/hadoop_catalog_test');
+'iceberg.catalog_location'='$WAREHOUSE_LOCATION_PREFIX/test-warehouse/$DATABASE/hadoop_catalog_test');
---- RESULTS
'Table has been created.'
====
@@ -171,13 +171,13 @@ PARTITIONED BY SPEC
)
STORED AS ICEBERG
TBLPROPERTIES('iceberg.catalog'='hadoop.catalog',
-'iceberg.catalog_location'='$WAREHOUSE_LOCATION_PREFIX/test_warehouse/$DATABASE/hadoop_catalog_test');
+'iceberg.catalog_location'='$WAREHOUSE_LOCATION_PREFIX/test-warehouse/$DATABASE/hadoop_catalog_test');
CREATE EXTERNAL TABLE iceberg_hadoop_cat_external(
level STRING
)
STORED AS ICEBERG
TBLPROPERTIES('iceberg.catalog'='hadoop.catalog',
-'iceberg.catalog_location'='$WAREHOUSE_LOCATION_PREFIX/test_warehouse/$DATABASE/hadoop_catalog_test', 'iceberg.table_identifier'='$DATABASE.iceberg_hadoop_catalog');
+'iceberg.catalog_location'='$WAREHOUSE_LOCATION_PREFIX/test-warehouse/$DATABASE/hadoop_catalog_test', 'iceberg.table_identifier'='$DATABASE.iceberg_hadoop_catalog');
---- RESULTS
'Table has been created.'
====
@@ -476,7 +476,7 @@ create table ice_part_hadoop_catalog (
col_identity
) stored as iceberg TBLPROPERTIES(
'iceberg.catalog' = 'hadoop.catalog',
- 'iceberg.catalog_location' = '$WAREHOUSE_LOCATION_PREFIX/test_warehouse/$DATABASE/hadoop_catalog_test'
+ 'iceberg.catalog_location' = '$WAREHOUSE_LOCATION_PREFIX/test-warehouse/$DATABASE/hadoop_catalog_test'
);
---- RESULTS
'Table has been created.'
diff --git a/testdata/workloads/functional-query/queries/QueryTest/iceberg-negative.test b/testdata/workloads/functional-query/queries/QueryTest/iceberg-negative.test
index 2474901ca..ff9f2b5aa 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/iceberg-negative.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/iceberg-negative.test
@@ -266,7 +266,7 @@ CREATE TABLE iceberg_table_hadoop_catalog(
)
STORED AS ICEBERG
TBLPROPERTIES('iceberg.catalog'='hadoop.catalog',
-'iceberg.catalog_location'='$WAREHOUSE_LOCATION_PREFIX/test_warehouse/$DATABASE/hadoop_catalog_test');
+'iceberg.catalog_location'='$WAREHOUSE_LOCATION_PREFIX/test-warehouse/$DATABASE/hadoop_catalog_test');
ALTER TABLE iceberg_table_hadoop_catalog RENAME TO iceberg_table_hadoop_catalog_new;
---- CATCH
UnsupportedOperationException: Cannot rename Iceberg tables that use 'hadoop.catalog' as catalog.
(impala) 03/03: IMPALA-13237: [Patch 6] - Match Behavior of Older Impala Versions
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jasonmfehr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 75ca356ed0451e1a98f70433a4d2729018b68d32
Author: jasonmfehr <jf...@cloudera.com>
AuthorDate: Tue Aug 12 14:30:16 2025 -0700
IMPALA-13237: [Patch 6] - Match Behavior of Older Impala Versions
Older versions of Impala set exec_state_ in ClientRequestState to
the error state earlier in the query processing than later versions.
This difference translates to when OpenTelemetry trace child spans
report an error status.
For example, in older Impala versions, if a query specifies a column
that does not exist, then the Planning child span has a status of
ERROR. However, in the latest version, the Planning span has a status of
OK, and only the Close span has a status of ERROR.
This difference caused the custom cluster test
test_otel_trace.py::TestOtelTrace::test_invalid_sql to fail in the
older Impala versions but pass in the latest version.
Additionally, older versions of Impala report a different default db.
The latest version reports whatever the client set. This difference
caused test_otel_trace.py::TestOtelTrace::test_retry_select_success
and test_otel_trace.py::TestOtelTrace::test_retry_select_failed to
fail in the older Impala versions because Impala used "tpch" as the
default db while the latest version used "default".
This change causes the OpenTelemetry trace child span where an error
actually occurs to report an error status, matching the behavior
of older Impala versions.
It also modifies test_otel_trace.py to expect the default db in the
OpenTelemetry trace "DefaultDb" attribute to match the query profile.
Testing accomplished by running the test_otel_trace.py custom cluster
tests.
Change-Id: If57aaef2da6d6904c66d0150f50ea9ac1c3ebc8c
Reviewed-on: http://gerrit.cloudera.org:8080/23293
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/observe/span-manager.cc | 7 ++-
tests/custom_cluster/test_otel_trace.py | 106 +++++++++++++++-----------------
tests/util/query_profile_util.py | 7 +++
3 files changed, 59 insertions(+), 61 deletions(-)
diff --git a/be/src/observe/span-manager.cc b/be/src/observe/span-manager.cc
index 3b4218937..38f42d820 100644
--- a/be/src/observe/span-manager.cc
+++ b/be/src/observe/span-manager.cc
@@ -403,9 +403,6 @@ void SpanManager::EndChildSpan(const Status* cause,
current_child_->SetAttribute(a.first, a.second);
}
- current_child_->SetAttribute(ATTR_STATUS,
- ClientRequestState::ExecStateToString(client_request_state_->exec_state()));
-
const Status* query_status;
if (cause != nullptr) {
query_status = cause;
@@ -415,6 +412,8 @@ void SpanManager::EndChildSpan(const Status* cause,
if (query_status->ok()) {
current_child_->SetAttributeEmpty(ATTR_ERROR_MSG);
+ current_child_->SetAttribute(ATTR_STATUS,
+ ClientRequestState::ExecStateToString(client_request_state_->exec_state()));
} else {
string error_msg = query_status->msg().msg();
@@ -423,6 +422,8 @@ void SpanManager::EndChildSpan(const Status* cause,
}
current_child_->SetAttribute(ATTR_ERROR_MSG, error_msg);
+ current_child_->SetAttribute(ATTR_STATUS,
+ ClientRequestState::ExecStateToString(ClientRequestState::ExecState::ERROR));
}
current_child_->End();
diff --git a/tests/custom_cluster/test_otel_trace.py b/tests/custom_cluster/test_otel_trace.py
index 0a948d7f1..25647011c 100644
--- a/tests/custom_cluster/test_otel_trace.py
+++ b/tests/custom_cluster/test_otel_trace.py
@@ -19,14 +19,13 @@ from __future__ import absolute_import, division, print_function
from tests.common.custom_cluster_test_suite import CustomClusterTestSuite
from tests.common.file_utils import wait_for_file_line_count
-from tests.common.impala_connection import ERROR, RUNNING, FINISHED
-from tests.common.test_vector import PROTOCOL, HS2, BEESWAX, ImpalaTestDimension
+from tests.common.impala_connection import ERROR, RUNNING, FINISHED, INITIALIZED
from tests.util.otel_trace import parse_trace_file, ATTR_VAL_TYPE_STRING, \
ATTR_VAL_TYPE_INT, ATTR_VAL_TYPE_BOOL
from tests.util.query_profile_util import parse_db_user, parse_session_id, parse_sql, \
parse_query_type, parse_query_status, parse_impala_query_state, parse_query_id, \
parse_retry_status, parse_original_query_id, parse_retried_query_id, \
- parse_num_rows_fetched, parse_admission_result
+ parse_num_rows_fetched, parse_admission_result, parse_default_db
from tests.util.retry import retry
@@ -36,11 +35,6 @@ class TestOtelTrace(CustomClusterTestSuite):
OUT_DIR = "out_dir"
TRACE_FILE = "export-trace.jsonl"
- @classmethod
- def add_test_dimensions(cls):
- super(TestOtelTrace, cls).add_test_dimensions()
- cls.ImpalaTestMatrix.add_dimension(ImpalaTestDimension(PROTOCOL, HS2, BEESWAX))
-
def setup_method(self, method):
super(TestOtelTrace, self).setup_method(method)
@@ -49,12 +43,11 @@ class TestOtelTrace(CustomClusterTestSuite):
"--otel_trace_exporter=file --otel_file_flush_interval_ms=500 "
"--otel_file_pattern={out_dir}/" + TRACE_FILE,
cluster_size=1, tmp_dir_placeholders=[OUT_DIR], disable_log_buffering=True)
- def test_query_success(self, vector):
+ def test_query_success(self):
"""Test that OpenTelemetry tracing is working by running a simple query and
checking that the trace file is created and contains spans."""
query = "select count(*) from functional.alltypes"
- result = self.execute_query_expect_success(
- self.create_impala_client_from_vector(vector), query)
+ result = self.execute_query_expect_success(self.client, query)
self.__assert_trace(result.query_id, result.runtime_profile, "otel_trace")
@@ -64,10 +57,9 @@ class TestOtelTrace(CustomClusterTestSuite):
"--otel_file_flush_interval_ms=500 "
"--otel_file_pattern={out_dir}/" + TRACE_FILE,
cluster_size=1, tmp_dir_placeholders=[OUT_DIR], disable_log_buffering=True)
- def test_invalid_sql(self, vector):
+ def test_invalid_sql(self):
query = "select * from functional.alltypes where field_does_not_exist=1"
- self.execute_query_expect_failure(self.create_impala_client_from_vector(vector),
- query)
+ self.execute_query_expect_failure(self.client, query)
# Retrieve the query id and runtime profile from the UI since the query execute call
# only returns a HiveServer2Error object and not the query id or profile.
@@ -90,12 +82,11 @@ class TestOtelTrace(CustomClusterTestSuite):
cluster_size=3, num_exclusive_coordinators=1, tmp_dir_placeholders=[OUT_DIR],
disable_log_buffering=True,
statestored_args="-statestore_heartbeat_frequency_ms=60000")
- def test_retry_select_success(self, vector):
+ def test_retry_select_success(self):
query = "select count(*) from tpch_parquet.lineitem where l_orderkey < 50"
self.cluster.impalads[1].kill()
- result = self.execute_query_expect_success(
- self.create_impala_client_from_vector(vector), query,
+ result = self.execute_query_expect_success(self.client, query,
{"RETRY_FAILED_QUERIES": True})
retried_query_id = parse_query_id(result.runtime_profile)
orig_query_profile = self.query_profile_from_ui(result.query_id)
@@ -124,36 +115,36 @@ class TestOtelTrace(CustomClusterTestSuite):
cluster_size=3, num_exclusive_coordinators=1, tmp_dir_placeholders=[OUT_DIR],
disable_log_buffering=True,
statestored_args="-statestore_heartbeat_frequency_ms=1000")
- def test_retry_select_failed(self, vector):
+ def test_retry_select_failed(self):
# Shuffle heavy query.
query = "select * from tpch.lineitem t1, tpch.lineitem t2 where " \
"t1.l_orderkey = t2.l_orderkey order by t1.l_orderkey, t2.l_orderkey limit 1"
- vector.set_exec_option("retry_failed_queries", "true")
- client = self.create_impala_client_from_vector(vector)
+ with self.create_impala_client() as client:
+ client.set_configuration({"retry_failed_queries": "true"})
- # Launch a query, it should be retried.
- handle = self.execute_query_async_using_client(client, query, vector)
- client.wait_for_impala_state(handle, RUNNING, 60)
- query_id = client.handle_id(handle)
- self.cluster.impalads[1].kill()
+ # Launch a query, it should be retried.
+ handle = client.execute_async(query)
+ client.wait_for_impala_state(handle, RUNNING, 60)
+ query_id = client.handle_id(handle)
+ self.cluster.impalads[1].kill()
- # Wait until the retry is running.
- def __wait_until_retried():
- return parse_retry_status(self.query_profile_from_ui(query_id)) == "RETRIED"
- retry(__wait_until_retried, 60, 1, 1, False)
+ # Wait until the retry is running.
+ def __wait_until_retried():
+ return parse_retry_status(self.query_profile_from_ui(query_id)) == "RETRIED"
+ retry(__wait_until_retried, 60, 1, 1, False)
- # Kill another impalad so that another retry is attempted.
- self.cluster.impalads[2].kill()
+ # Kill another impalad so that another retry is attempted.
+ self.cluster.impalads[2].kill()
- # Wait until the query fails.
- client.wait_for_impala_state(handle, ERROR, 60)
+ # Wait until the query fails.
+ client.wait_for_impala_state(handle, ERROR, 60)
- retried_query_profile = client.get_runtime_profile(handle)
- retried_query_id = parse_query_id(retried_query_profile)
- orig_query_profile = self.query_profile_from_ui(query_id)
+ retried_query_profile = client.get_runtime_profile(handle)
+ retried_query_id = parse_query_id(retried_query_profile)
+ orig_query_profile = self.query_profile_from_ui(query_id)
- client.close_query(handle)
+ client.close_query(handle)
# Assert the trace from the original query.
self.__assert_trace(
@@ -179,27 +170,24 @@ class TestOtelTrace(CustomClusterTestSuite):
"--otel_file_pattern={out_dir}/" + TRACE_FILE + " "
"--default_pool_max_requests=1",
cluster_size=1, tmp_dir_placeholders=[OUT_DIR], disable_log_buffering=True)
- def test_select_queued(self, vector):
+ def test_select_queued(self):
# Launch two queries, the second will be queued until the first completes.
- client = self.create_impala_client_from_vector(vector)
-
query = "select * from functional.alltypes where id = 1"
- handle1 = self.execute_query_async_using_client(client,
- "{} and int_col = sleep(5000)".format(query), vector)
- client.wait_for_impala_state(handle1, RUNNING, 60)
- query_id_1 = client.handle_id(handle1)
+ handle1 = self.client.execute_async("{} and int_col = sleep(5000)".format(query))
+ self.client.wait_for_impala_state(handle1, RUNNING, 60)
+ query_id_1 = self.client.handle_id(handle1)
- handle2 = self.execute_query_async_using_client(client, query, vector)
- query_id_2 = client.handle_id(handle2)
+ handle2 = self.client.execute_async(query)
+ query_id_2 = self.client.handle_id(handle2)
- client.wait_for_impala_state(handle1, FINISHED, 60)
- query_profile_1 = client.get_runtime_profile(handle1)
- client.close_query(handle1)
+ self.client.wait_for_impala_state(handle1, FINISHED, 60)
+ query_profile_1 = self.client.get_runtime_profile(handle1)
+ self.client.close_query(handle1)
- client.wait_for_impala_state(handle2, FINISHED, 60)
- query_profile_2 = client.get_runtime_profile(handle2)
+ self.client.wait_for_impala_state(handle2, FINISHED, 60)
+ query_profile_2 = self.client.get_runtime_profile(handle2)
- client.close_query(handle2)
+ self.client.close_query(handle2)
self.__assert_trace(
query_id=query_id_1,
@@ -270,8 +258,8 @@ class TestOtelTrace(CustomClusterTestSuite):
span_err_msg = query_status
in_error = True
self.__assert_initspan_attrs(trace.child_spans, root_span_id, query_id, session_id,
- cluster_id, db_user, "default-pool", "default", parse_sql(query_profile),
- original_query_id)
+ cluster_id, db_user, "default-pool", parse_default_db(query_profile),
+ parse_sql(query_profile), original_query_id)
# Assert Submitted span.
if "Submitted" not in missing_spans:
@@ -283,15 +271,17 @@ class TestOtelTrace(CustomClusterTestSuite):
# Assert Planning span.
if "Planning" not in missing_spans:
+ status = INITIALIZED
span_err_msg = ""
if err_span == "Planning" or in_error:
span_err_msg = query_status
+ status = ERROR
in_error = True
query_type = parse_query_type(query_profile)
if query_type == "N/A":
query_type = "UNKNOWN"
self.__assert_planningspan_attrs(trace.child_spans, root_span_id, query_id,
- query_type, span_err_msg)
+ query_type, span_err_msg, status)
# Assert AdmissionControl span.
if "AdmissionControl" not in missing_spans:
@@ -487,7 +477,7 @@ class TestOtelTrace(CustomClusterTestSuite):
# Locate the init span and assert.
init_span = self.__find_span(spans, "Init", query_id)
- self.__assert_scopespan_common(init_span, query_id, False, "Init", 8, "INITIALIZED",
+ self.__assert_scopespan_common(init_span, query_id, False, "Init", 8, INITIALIZED,
root_span_id)
self.__assert_attr(init_span.name, init_span.attributes, "QueryId", query_id)
@@ -507,10 +497,10 @@ class TestOtelTrace(CustomClusterTestSuite):
submitted_span = self.__find_span(spans, "Submitted", query_id)
self.__assert_scopespan_common(submitted_span, query_id, False, "Submitted", 0,
- "INITIALIZED", root_span_id)
+ INITIALIZED, root_span_id)
def __assert_planningspan_attrs(self, spans, root_span_id, query_id, query_type,
- err_msg="", status="INITIALIZED"):
+ err_msg="", status=INITIALIZED):
"""
Helper function that asserts the common and span-specific attributes in the
planning execution span.
diff --git a/tests/util/query_profile_util.py b/tests/util/query_profile_util.py
index 04e76e3d3..2b9827ed7 100644
--- a/tests/util/query_profile_util.py
+++ b/tests/util/query_profile_util.py
@@ -120,3 +120,10 @@ def parse_admission_result(profile_text):
admission_result = re.search(r'\n\s+Admission result:\s+(.*?)\n', profile_text)
assert admission_result is not None, "Admission Result not found in query profile"
return admission_result.group(1)
+
+
+def parse_default_db(profile_text):
+ """Parses the default db from the query profile text."""
+ default_db = re.search(r'\n\s+Default Db:\s+(.*?)\n', profile_text)
+ assert default_db is not None, "Default Db not found in query profile"
+ return default_db.group(1)