You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@linkis.apache.org by ki...@apache.org on 2025/11/18 13:04:15 UTC
(linkis) 03/03: fix hive concurrent mode load udf error problems
This is an automated email from the ASF dual-hosted git repository.
kinghao pushed a commit to branch 1.9.0-fixhivecurrenc
in repository https://gitbox.apache.org/repos/asf/linkis.git
commit 55c9df0c401223b9614cc04337e28ad93c17fdd7
Author: kinghao <ki...@apache.org>
AuthorDate: Tue Nov 18 21:03:34 2025 +0800
fix hive concurrent mode load udf error problems
---
.../common/conf/EnvConfiguration.scala | 2 --
.../entrance/interceptor/impl/SQLExplainTest.java | 24 ++++++++++------------
.../HiveEngineConcurrentConnExecutor.scala | 9 ++------
.../hive/hook/HiveAddJarsEngineHook.scala | 8 +++++++-
4 files changed, 20 insertions(+), 23 deletions(-)
diff --git a/linkis-computation-governance/linkis-engineconn/linkis-engineconn-plugin-core/src/main/scala/org/apache/linkis/manager/engineplugin/common/conf/EnvConfiguration.scala b/linkis-computation-governance/linkis-engineconn/linkis-engineconn-plugin-core/src/main/scala/org/apache/linkis/manager/engineplugin/common/conf/EnvConfiguration.scala
index 7689ae94af..5a85334f19 100644
--- a/linkis-computation-governance/linkis-engineconn/linkis-engineconn-plugin-core/src/main/scala/org/apache/linkis/manager/engineplugin/common/conf/EnvConfiguration.scala
+++ b/linkis-computation-governance/linkis-engineconn/linkis-engineconn-plugin-core/src/main/scala/org/apache/linkis/manager/engineplugin/common/conf/EnvConfiguration.scala
@@ -19,8 +19,6 @@ package org.apache.linkis.manager.engineplugin.common.conf
import org.apache.linkis.common.conf.{ByteType, CommonVars, Configuration}
-import org.apache.commons.lang3.{JavaVersion, SystemUtils}
-
object EnvConfiguration {
val HIVE_CONF_DIR = CommonVars[String](
diff --git a/linkis-computation-governance/linkis-entrance/src/test/java/org/apache/linkis/entrance/interceptor/impl/SQLExplainTest.java b/linkis-computation-governance/linkis-entrance/src/test/java/org/apache/linkis/entrance/interceptor/impl/SQLExplainTest.java
index 49e65e964a..94f2a9b4d4 100644
--- a/linkis-computation-governance/linkis-entrance/src/test/java/org/apache/linkis/entrance/interceptor/impl/SQLExplainTest.java
+++ b/linkis-computation-governance/linkis-entrance/src/test/java/org/apache/linkis/entrance/interceptor/impl/SQLExplainTest.java
@@ -18,6 +18,7 @@
package org.apache.linkis.entrance.interceptor.impl;
import org.apache.linkis.governance.common.entity.job.JobRequest;
+
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -55,24 +56,21 @@ class SQLExplainTest {
}
/**
- * 未修复前代码进行拼接sql时，输出的sql为
- * select
- * id,
- * name,
- * array_join(array_intersect(map_keys(info),array['abs','oda'],' limit 5000;
- * ') as infos
- * from ods.dim_ep22
+ * 未修复前代码进行拼接sql时，输出的sql为 select id, name,
+ * array_join(array_intersect(map_keys(info),array['abs','oda'],' limit 5000; ') as infos from
+ * ods.dim_ep22
*/
@Test
void splicingLimitSql() {
- String code = "select\n" +
- "id,\n" +
- "name,\n" +
- "array_join(array_intersect(map_keys(info),array['abs','oda'],';') as infos\n" +
- "from ods.dim_ep22";
+ String code =
+ "select\n"
+ + "id,\n"
+ + "name,\n"
+ + "array_join(array_intersect(map_keys(info),array['abs','oda'],';') as infos\n"
+ + "from ods.dim_ep22";
StringBuilder logAppender = new StringBuilder();
JobRequest jobRequest = new JobRequest();
SQLExplain.dealSQLLimit(code, jobRequest, logAppender);
- Assertions.assertEquals(code+" limit 5000", jobRequest.getExecutionCode());
+ Assertions.assertEquals(code + " limit 5000", jobRequest.getExecutionCode());
}
}
diff --git a/linkis-engineconn-plugins/hive/src/main/scala/org/apache/linkis/engineplugin/hive/executor/HiveEngineConcurrentConnExecutor.scala b/linkis-engineconn-plugins/hive/src/main/scala/org/apache/linkis/engineplugin/hive/executor/HiveEngineConcurrentConnExecutor.scala
index 47496cf17a..9fe37eb3d9 100644
--- a/linkis-engineconn-plugins/hive/src/main/scala/org/apache/linkis/engineplugin/hive/executor/HiveEngineConcurrentConnExecutor.scala
+++ b/linkis-engineconn-plugins/hive/src/main/scala/org/apache/linkis/engineplugin/hive/executor/HiveEngineConcurrentConnExecutor.scala
@@ -135,7 +135,7 @@ class HiveEngineConcurrentConnExecutor(
code: String
): ExecuteResponse = {
LOG.info(s"HiveEngineConcurrentConnExecutor Ready to executeLine: $code")
- val taskId: String = engineExecutorContext.getJobId.get
+ val taskId: String = engineExecutorContext.getJobId.getOrElse("udf_init")
CSHiveHelper.setContextIDInfoToHiveConf(engineExecutorContext, hiveConf)
val realCode = code.trim()
@@ -166,12 +166,7 @@ class HiveEngineConcurrentConnExecutor(
val driver = new HiveDriverProxy(any)
driverCache.put(taskId, driver)
- executeHQL(
- engineExecutorContext.getJobId.get,
- engineExecutorContext,
- realCode,
- driver
- )
+ executeHQL(taskId, engineExecutorContext, realCode, driver)
case _ =>
val resp = proc.run(realCode.substring(tokens(0).length).trim)
val result = new String(baos.toByteArray)
diff --git a/linkis-engineconn-plugins/hive/src/main/scala/org/apache/linkis/engineplugin/hive/hook/HiveAddJarsEngineHook.scala b/linkis-engineconn-plugins/hive/src/main/scala/org/apache/linkis/engineplugin/hive/hook/HiveAddJarsEngineHook.scala
index 9019db9621..cd94df7f73 100644
--- a/linkis-engineconn-plugins/hive/src/main/scala/org/apache/linkis/engineplugin/hive/hook/HiveAddJarsEngineHook.scala
+++ b/linkis-engineconn-plugins/hive/src/main/scala/org/apache/linkis/engineplugin/hive/hook/HiveAddJarsEngineHook.scala
@@ -24,7 +24,10 @@ import org.apache.linkis.engineconn.common.engineconn.EngineConn
import org.apache.linkis.engineconn.common.hook.EngineConnHook
import org.apache.linkis.engineconn.computation.executor.execute.EngineExecutionContext
import org.apache.linkis.engineconn.core.executor.ExecutorManager
-import org.apache.linkis.engineplugin.hive.executor.HiveEngineConnExecutor
+import org.apache.linkis.engineplugin.hive.executor.{
+ HiveEngineConcurrentConnExecutor,
+ HiveEngineConnExecutor
+}
import org.apache.linkis.manager.engineplugin.common.launch.process.Environment
import org.apache.linkis.manager.label.entity.Label
import org.apache.linkis.manager.label.entity.engine.{CodeLanguageLabel, RunType}
@@ -78,7 +81,10 @@ class HiveAddJarsEngineHook extends EngineConnHook with Logging {
ExecutorManager.getInstance.getExecutorByLabels(labels) match {
case executor: HiveEngineConnExecutor =>
executor.executeLine(new EngineExecutionContext(executor), sql)
+ case executor: HiveEngineConcurrentConnExecutor =>
+ executor.executeLine(new EngineExecutionContext(executor), sql)
case _ =>
+ logger.warn(s"Executor is not a ComputationExecutor, skip adding jar: $jar")
}
} catch {
case t: Throwable => logger.error(s"run hive sql ${addSql + jar} failed", t)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@linkis.apache.org
For additional commands, e-mail: commits-help@linkis.apache.org