You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by ar...@apache.org on 2025/03/05 20:52:47 UTC
(impala) branch master updated (24179d46c -> 28686165d)
This is an automated email from the ASF dual-hosted git repository.
arawat pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
from 24179d46c IMPALA-13793: Always close expressions
new ea989dfb2 IMPALA-13815: Fix Flaky Workload Management Tests
new 28686165d IMPALA-13817: Impala fails to start if 'ai_endpoint' and 'ai_additional_platforms' are not set in the right order
The 2 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/exprs/ai-functions-ir.cc | 4 +--
be/src/exprs/ai-functions.cc | 3 +--
be/src/exprs/ai-functions.h | 2 ++
be/src/exprs/expr-test.cc | 15 +++++++++++
be/src/runtime/exec-env.cc | 13 ++++++++++
tests/common/custom_cluster_test_suite.py | 42 ++++++++++++++-----------------
tests/custom_cluster/test_query_live.py | 8 ------
7 files changed, 52 insertions(+), 35 deletions(-)
(impala) 01/02: IMPALA-13815: Fix Flaky Workload Management Tests
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arawat pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit ea989dfb283d1c80d96066f163d3f709adcf411d
Author: jasonmfehr <jf...@cloudera.com>
AuthorDate: Mon Mar 3 15:52:51 2025 -0800
IMPALA-13815: Fix Flaky Workload Management Tests
The CustomClusterTestSuite.wait_for_wm_init() function checks for two
specific log lines to be logged by the catalog. The first line is
logged when workload manangement initialization is complete. The
second line is when a catalog topic update has been assembled.
However, if workload management initialization is slow, then there
may not be a catalog topic update assembled after the initialization
completes. When this happens, an assertion fails despite the workload
management tables having been properly initialized and loaded by the
catalog.
This patch simplifies the CustomClusterTestSuite.wait_for_wm_init()
function so it waits until the catalogd logs it has completed
workload management initialization and then checks each coordinator's
local catalog cache for the workload management tables.
The following test suites passed locally and in an ASAN build. These
tests all call the 'wait_for_wm_init' function of
CustomClusterTestSuite.
* tests/custom_cluster/test_query_live.py
* tests/custom_cluster/test_query_log.py
* tests/custom_cluster/test_workload_mgmt_init.py
* tests/custom_cluster/test_workload_mgmt_sql_details.py
Change-Id: Ieb4c86fa79bb1df000b6241bdd31c7641d807c4f
Reviewed-on: http://gerrit.cloudera.org:8080/22570
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Riza Suminto <ri...@cloudera.com>
---
tests/common/custom_cluster_test_suite.py | 42 ++++++++++++++-----------------
tests/custom_cluster/test_query_live.py | 8 ------
2 files changed, 19 insertions(+), 31 deletions(-)
diff --git a/tests/common/custom_cluster_test_suite.py b/tests/common/custom_cluster_test_suite.py
index ae01adfb6..8819b8735 100644
--- a/tests/common/custom_cluster_test_suite.py
+++ b/tests/common/custom_cluster_test_suite.py
@@ -37,6 +37,7 @@ from tests.common.impala_test_suite import ImpalaTestSuite
from tests.common.impala_cluster import ImpalaCluster
from tests.util.filesystem_utils import IS_LOCAL
from tests.util.retry import retry
+from tests.util.workload_management import QUERY_TBL_LOG_NAME, QUERY_TBL_LIVE_NAME
from time import sleep
LOG = logging.getLogger(__name__)
@@ -410,30 +411,25 @@ class CustomClusterTestSuite(ImpalaTestSuite):
def wait_for_wm_init_complete(self, timeout_s=60):
"""
Waits for the catalog to report the workload management initialization process has
- completed and for the catalog updates to be received by the coordinators. The input
- timeout_s is used as the timeout for three separate function calls. Thus, the
- theoretical max amount of time this function could wait is (timeout_s * 3).
+ completed and the workload management tables to be in the local catalog. The input
+ timeout_s is used as the timeout for multiple separate function calls. Thus, the
+ theoretical max amount of time this function could wait is:
+ timeout_s + (timeout_s * num_coordinators).
"""
- catalog_log = self.assert_log_contains_multiline("catalogd", "INFO", r'Completed '
- r'workload management initialization.*?A catalog update with \d+ entries is '
- r'assembled\. Catalog version: (\d+)', timeout_s)
-
- # Assert each coordinator has received a catalog update that was assembled after
- # workload management completed.
- for idx, _ in enumerate(self.cluster.get_all_coordinators()):
- node_name = "impalad"
- if idx > 0:
- node_name += "_node" + str(idx)
-
- def assert_func():
- coord_log = self.assert_log_contains(node_name, "INFO", r'Catalog topic update '
- r'applied with version: (\d+)', timeout_s=timeout_s, expected_count=-1)
- return int(coord_log.group(1)) >= int(catalog_log.group(1))
-
- max_attempts = timeout_s / 3
- assert retry(func=assert_func, max_attempts=max_attempts, sleep_time_s=3,
- backoff=1), "Expected a catalog topic update with version '{}' or later, but " \
- "no such update was found.".format(catalog_log.group(1))
+ self.assert_catalogd_ha_contains("INFO", r'Completed workload management '
+ r'initialization', timeout_s)
+
+ for tbl in (QUERY_TBL_LIVE_NAME, QUERY_TBL_LOG_NAME):
+ for coord in self.cluster.get_all_coordinators():
+ # Wait until table is available in the coordinator's catalog cache.
+ def exists_func():
+ catalog_objs = coord.service.read_debug_webpage("catalog?json")
+ return tbl in catalog_objs
+
+ max_attempts = timeout_s / 3
+ assert retry(func=exists_func, max_attempts=max_attempts, sleep_time_s=3,
+ backoff=1), "Did not find table '{}' in local catalog of coordinator " \
+ "'{}:{}'.".format(tbl, coord.hostname, coord.get_webserver_port())
@classmethod
def _stop_impala_cluster(cls):
diff --git a/tests/custom_cluster/test_query_live.py b/tests/custom_cluster/test_query_live.py
index 3175e544a..e12254a27 100644
--- a/tests/custom_cluster/test_query_live.py
+++ b/tests/custom_cluster/test_query_live.py
@@ -36,14 +36,6 @@ class TestQueryLive(CustomClusterTestSuite):
super(TestQueryLive, self).setup_method(method)
self.wait_for_wm_init_complete()
- # Wait until sys.impala_query_live is available in the coordinator's catalog cache.
- def table_exists():
- catalog_objs = self.cluster.get_first_impalad() \
- .service.read_debug_webpage("catalog?json")
- return "impala_query_live" in catalog_objs
-
- assert retry(func=table_exists, max_attempts=5, sleep_time_s=3, backoff=1)
-
def assert_describe_extended(self):
describe_ext_result = self.execute_query('describe extended sys.impala_query_live')
# Alter can add additional event fields. Filter them out.
(impala) 02/02: IMPALA-13817: Impala fails to start if 'ai_endpoint' and 'ai_additional_platforms' are not set in the right order
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arawat pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 28686165d089bef85c76fce31ac1135e508b1cb2
Author: Abhishek Rawat <ar...@cloudera.com>
AuthorDate: Sun Mar 2 21:23:51 2025 -0800
IMPALA-13817: Impala fails to start if 'ai_endpoint' and
'ai_additional_platforms' are not set in the right order
The 'ai_endpoint' flag has a validator which depends on
'ai_additional_platforms'. This could result in failure if the
'ai_additional_platforms' is not set before 'ai_endpoint' in the
flagfile.
This patch simplifies the validator for 'ai_endpoint' flag and moves
the supported platforms check to ExecEnv::Init.
Testing:
- Added unit tests for 'is_api_endpoint_supported' function.
Change-Id: I16480564ce434513ab73b531c03e1d229dac2dd6
Reviewed-on: http://gerrit.cloudera.org:8080/22567
Reviewed-by: Yida Wu <wy...@gmail.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/exprs/ai-functions-ir.cc | 4 ++--
be/src/exprs/ai-functions.cc | 3 +--
be/src/exprs/ai-functions.h | 2 ++
be/src/exprs/expr-test.cc | 15 +++++++++++++++
be/src/runtime/exec-env.cc | 13 +++++++++++++
5 files changed, 33 insertions(+), 4 deletions(-)
diff --git a/be/src/exprs/ai-functions-ir.cc b/be/src/exprs/ai-functions-ir.cc
index a3a40459d..2b36e268b 100644
--- a/be/src/exprs/ai-functions-ir.cc
+++ b/be/src/exprs/ai-functions-ir.cc
@@ -62,8 +62,8 @@ const char* AiFunctions::AZURE_OPEN_AI_REQUEST_AUTH_HEADER =
// other constants
static const StringVal NULL_STRINGVAL = StringVal::null();
static const char* AI_API_ENDPOINT_PREFIX = "https://";
-static const char* OPEN_AI_AZURE_ENDPOINT = "openai.azure.com";
-static const char* OPEN_AI_PUBLIC_ENDPOINT = "api.openai.com";
+const char* AiFunctions::OPEN_AI_AZURE_ENDPOINT = "openai.azure.com";
+const char* AiFunctions::OPEN_AI_PUBLIC_ENDPOINT = "api.openai.com";
// OPEN AI specific constants
static const char* OPEN_AI_RESPONSE_FIELD_CHOICES = "choices";
static const char* OPEN_AI_RESPONSE_FIELD_MESSAGE = "message";
diff --git a/be/src/exprs/ai-functions.cc b/be/src/exprs/ai-functions.cc
index a9dadf844..2e7dd6617 100644
--- a/be/src/exprs/ai-functions.cc
+++ b/be/src/exprs/ai-functions.cc
@@ -37,8 +37,7 @@ using namespace impala_udf;
DEFINE_string(ai_endpoint, "https://api.openai.com/v1/chat/completions",
"The default API endpoint for an external AI engine.");
DEFINE_validator(ai_endpoint, [](const char* name, const string& endpoint) {
- return (impala::AiFunctions::is_api_endpoint_valid(endpoint) &&
- impala::AiFunctions::is_api_endpoint_supported(endpoint));
+ return impala::AiFunctions::is_api_endpoint_valid(endpoint);
});
DEFINE_string(ai_model, "gpt-4", "The default AI model used by an external AI engine.");
diff --git a/be/src/exprs/ai-functions.h b/be/src/exprs/ai-functions.h
index d8480a93b..d72c3b54c 100644
--- a/be/src/exprs/ai-functions.h
+++ b/be/src/exprs/ai-functions.h
@@ -39,6 +39,8 @@ class AiFunctions {
static const char* OPEN_AI_REQUEST_FIELD_CONTENT_TYPE_HEADER;
static const char* OPEN_AI_REQUEST_AUTH_HEADER;
static const char* AZURE_OPEN_AI_REQUEST_AUTH_HEADER;
+ static const char* OPEN_AI_AZURE_ENDPOINT;
+ static const char* OPEN_AI_PUBLIC_ENDPOINT;
enum class AI_PLATFORM {
/// Unsupported platform
UNSUPPORTED,
diff --git a/be/src/exprs/expr-test.cc b/be/src/exprs/expr-test.cc
index 21be6c6a5..0662354c3 100644
--- a/be/src/exprs/expr-test.cc
+++ b/be/src/exprs/expr-test.cc
@@ -11598,6 +11598,21 @@ TEST_P(ExprTest, AiFunctionsTestAdditionalSites) {
EXPECT_EQ(
AiFunctions::GetAiPlatformFromEndpoint("https://AI-API.COM/v1/generate", true),
AiFunctions::AI_PLATFORM::GENERAL);
+
+ // Test for supported/unsupported endpoints.
+ EXPECT_EQ(
+ AiFunctions::is_api_endpoint_supported("https://ai-api.com/v1/generate"), true);
+ EXPECT_EQ(
+ AiFunctions::is_api_endpoint_supported("https://another-ai.org/completions"), true);
+ EXPECT_EQ(
+ AiFunctions::is_api_endpoint_supported(
+ "https://api.openai.com/v1/chat/completions"), true);
+ EXPECT_EQ(
+ AiFunctions::is_api_endpoint_supported(
+ "https://openai.azure.com/openai/deployments/"), true);
+ EXPECT_EQ(
+ AiFunctions::is_api_endpoint_supported("https://random-ai.com/generate"), false);
+
}
} // namespace impala
diff --git a/be/src/runtime/exec-env.cc b/be/src/runtime/exec-env.cc
index eb614421e..bf45e42bb 100644
--- a/be/src/runtime/exec-env.cc
+++ b/be/src/runtime/exec-env.cc
@@ -157,6 +157,8 @@ DECLARE_string(debug_actions);
DECLARE_string(ssl_client_ca_certificate);
DECLARE_string(ai_api_key_jceks_secret);
+DECLARE_string(ai_endpoint);
+DECLARE_string(ai_additional_platforms);
DEFINE_int32(backend_client_connection_num_retries, 3, "Retry backend connections.");
// When network is unstable, TCP will retry and sending could take longer time.
@@ -535,6 +537,17 @@ Status ExecEnv::Init() {
frontend_->GetSecretFromKeyStore(FLAGS_ai_api_key_jceks_secret, &api_key));
AiFunctions::set_api_key(api_key);
}
+ // Validate default ai_endpoint.
+ if (FLAGS_ai_endpoint != "" &&
+ !AiFunctions::is_api_endpoint_supported(FLAGS_ai_endpoint)) {
+ string supported_platforms = Substitute("$0, $1",
+ AiFunctions::OPEN_AI_AZURE_ENDPOINT, AiFunctions::OPEN_AI_PUBLIC_ENDPOINT);
+ if (!FLAGS_ai_additional_platforms.empty()) {
+ supported_platforms += ", " + FLAGS_ai_additional_platforms;
+ }
+ return Status(Substitute("Unsupported --ai_endpoint=$0, supported platforms are: $1",
+ FLAGS_ai_endpoint, supported_platforms));
+ }
jwt_helper_ = new JWTHelper();
oauth_helper_ = new JWTHelper();