You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@linkis.apache.org by pe...@apache.org on 2023/10/09 07:06:00 UTC
[linkis] branch master updated (16183c973 -> 9f1dd0911)
This is an automated email from the ASF dual-hosted git repository.
peacewong pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/linkis.git
from 16183c973 [Feature][linkis-public-enhancements] Mer pes publicservice (#4854)
new fcd6c7c0b 1. add result set sot utils class 2. optimize variable operation logic
new 2b6d80240 add default retry for 401
new d24599497 add debug switch params to task
new 50254ac55 optimize consumer code logic
new 01ae62a56 add ps datasource server conf
new dc8a7ad3c optimize storage module
new efc85b380 add version info to linkis-cli
new c625de68e optimize jdbc-driver module
new 88d9e1be7 support null
new b63f34c5b Fix build error
new 9f1dd0911 Fix build error
The 11 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:
.../java/org/apache/linkis/common/io/FsPath.java | 6 +
.../apache/linkis/common/utils/ByteTimeUtils.java | 13 +-
.../apache/linkis/common/utils/ResultSetUtils.java | 57 ++++++++
.../common/utils/VariableOperationUtils.java | 43 ++++--
.../apache/linkis/common/utils/VariableUtils.scala | 24 +++-
.../linkis/common/utils/ByteTimeUtilsTest.java | 152 +++++++++++++++++++--
.../linkis/common/utils/VariableUtilsTest.scala | 17 +++
.../linkis/httpclient/AbstractHttpClient.scala | 26 ++--
.../httpclient/config/ClientConfigBuilder.scala | 9 +-
linkis-commons/linkis-mybatis/pom.xml | 26 ++++
.../linkis/protocol/constants/TaskConstant.java | 1 +
.../apache/linkis/protocol/utils/TaskUtils.scala | 11 ++
.../apache/linkis/rpc/conf/RPCConfiguration.scala | 10 ++
.../linkis/rpc/conf/RPCConfigurationTest.scala | 2 -
.../apache/linkis/scheduler/queue/Consumer.scala | 1 -
.../linkis/scheduler/queue/GroupFactory.scala | 5 +
.../linkis/scheduler/queue/LoopArrayQueue.scala | 7 +-
.../queue/fifoqueue/FIFOUserConsumer.scala | 45 +++++-
.../linkis/storage/conf/LinkisStorageConf.java | 2 +-
.../linkis/storage/csv/StorageCSVWriter.java | 15 +-
.../org/apache/linkis/storage/domain/DataType.java | 15 +-
.../org/apache/linkis/storage/domain/Dolphin.java | 11 ++
.../linkis/storage/exception/StorageErrorCode.java | 5 +-
.../storage/resultset/ResultSetReaderFactory.java | 1 +
.../storage/resultset/StorageResultSetReader.java | 27 ++--
.../storage/resultset/table/TableRecord.java | 9 --
.../resultset/table/TableResultDeserializer.java | 6 +-
.../resultset/table/TableResultSerializer.java | 2 +-
.../apache/linkis/storage/source/FileSource.java | 1 +
.../linkis/storage/source/ResultsetFileSource.java | 14 +-
.../linkis-client/linkis-cli/pom.xml | 6 +
.../apache/linkis/cli/application/CtxBuilder.java | 21 ++-
.../cli/application/LinkisClientApplication.java | 41 ++++--
.../linkis/cli/application/constants/CliKeys.java | 2 +
.../cli/application/constants/LinkisKeys.java | 1 +
.../cli/application/entity/version/Version.java | 20 ---
.../command/template/UniversalCmdTemplate.java | 12 ++
.../application/interactor/job/help/HelpJob.java | 62 +++++++++
.../job/version/VersionJob.java} | 35 +++--
.../VersionJobResult.java} | 24 +---
.../application/operator/ujes/LinkisJobOper.java | 4 +-
.../resources/version.properties} | 6 +-
.../ujes/client/request/ResultSetAction.scala | 9 ++
.../org/apache/linkis/ujes/jdbc/UJESSQLDriver.java | 4 +-
.../apache/linkis/ujes/jdbc/utils/JDBCUtils.java | 10 ++
.../linkis/ujes/jdbc/LinkisSQLConnection.scala | 19 +++
.../linkis/ujes/jdbc/UJESClientFactory.scala | 1 -
.../linkis/ujes/jdbc/UJESSQLDriverMain.scala | 24 ++--
.../apache/linkis/ujes/jdbc/UJESSQLResultSet.scala | 14 +-
.../jdbc/hook/JDBCDriverPreExecutionHook.scala | 3 +-
.../ujes/jdbc/hook/impl/NoLimitExecutionHook.scala | 42 ------
51 files changed, 697 insertions(+), 226 deletions(-)
create mode 100644 linkis-commons/linkis-common/src/main/java/org/apache/linkis/common/utils/ResultSetUtils.java
delete mode 100644 linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/entity/version/Version.java
create mode 100644 linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/interactor/job/help/HelpJob.java
copy linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/{operator/ujes/LinkisOperBuilder.java => interactor/job/version/VersionJob.java} (54%)
copy linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/interactor/job/{common/KillResult.java => version/VersionJobResult.java} (67%)
copy linkis-computation-governance/linkis-client/linkis-cli/src/{test/resources/conf/user.properties => main/resources/version.properties} (94%)
delete mode 100644 linkis-computation-governance/linkis-jdbc-driver/src/main/scala/org/apache/linkis/ujes/jdbc/hook/impl/NoLimitExecutionHook.scala
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@linkis.apache.org
For additional commands, e-mail: commits-help@linkis.apache.org
[linkis] 11/11: Fix build error
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
peacewong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/linkis.git
commit 9f1dd091197d8e1c08459e0446ccfcf8bf5e25c1
Author: peacewong <wp...@gmail.com>
AuthorDate: Sat Oct 7 10:38:13 2023 +0800
Fix build error
---
.../ujes/jdbc/hook/impl/NoLimitExecutionHook.scala | 42 ----------------------
1 file changed, 42 deletions(-)
diff --git a/linkis-computation-governance/linkis-jdbc-driver/src/main/scala/org/apache/linkis/ujes/jdbc/hook/impl/NoLimitExecutionHook.scala b/linkis-computation-governance/linkis-jdbc-driver/src/main/scala/org/apache/linkis/ujes/jdbc/hook/impl/NoLimitExecutionHook.scala
deleted file mode 100644
index 103a54477..000000000
--- a/linkis-computation-governance/linkis-jdbc-driver/src/main/scala/org/apache/linkis/ujes/jdbc/hook/impl/NoLimitExecutionHook.scala
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.linkis.ujes.jdbc.hook.impl
-
-import org.apache.linkis.ujes.jdbc.UJESSQLDriverMain
-import org.apache.linkis.ujes.jdbc.hook.JDBCDriverPreExecutionHook
-
-import java.util.Locale
-
-class NoLimitExecutionHook extends JDBCDriverPreExecutionHook {
-
- override def callPreExecutionHook(sql: String, skip: Boolean): String = {
- if (UJESSQLDriverMain.LIMIT_ENABLED.equalsIgnoreCase("false")) {
- var noLimitSql = "--set ide.engine.no.limit.allow=true\n" + sql
- val lowerCaseLimitSql = noLimitSql.toLowerCase()
- if (lowerCaseLimitSql.contains("limit ") && lowerCaseLimitSql.contains("tableausql")) {
- val lastIndexOfLimit = lowerCaseLimitSql.lastIndexOf("limit ")
- noLimitSql = noLimitSql.substring(0, lastIndexOfLimit)
- }
- noLimitSql
- } else {
- sql
- }
-
- }
-
-}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@linkis.apache.org
For additional commands, e-mail: commits-help@linkis.apache.org
[linkis] 10/11: Fix build error
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
peacewong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/linkis.git
commit b63f34c5b2fa88db1880cd32eaab3615704b7de3
Author: peacewong <wp...@gmail.com>
AuthorDate: Wed Oct 4 23:49:50 2023 +0800
Fix build error
---
.../org/apache/linkis/storage/resultset/StorageResultSetReader.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/resultset/StorageResultSetReader.java b/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/resultset/StorageResultSetReader.java
index 23d65e061..c0222cc84 100644
--- a/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/resultset/StorageResultSetReader.java
+++ b/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/resultset/StorageResultSetReader.java
@@ -86,7 +86,7 @@ public class StorageResultSetReader<K extends MetaData, V extends Record>
byte[] rowBuffer = null;
try {
rowBuffer = new byte[rowLen];
- len = StorageUtils.readBytes(inputStream, rowBuffer, READ_CACHE);
+ len = StorageUtils.readBytes(inputStream, rowBuffer, rowLen);
} catch (OutOfMemoryError error) {
logger.error("Result set read oom, read size {} Byte", rowLen);
throw new RuntimeException(error);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@linkis.apache.org
For additional commands, e-mail: commits-help@linkis.apache.org
[linkis] 04/11: optimize consumer code logic
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
peacewong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/linkis.git
commit 50254ac558e1ab26321a67411b6baee402dece57
Author: peacewong <wp...@gmail.com>
AuthorDate: Fri Sep 22 14:55:54 2023 +0800
optimize consumer code logic
---
.../apache/linkis/scheduler/queue/Consumer.scala | 1 -
.../linkis/scheduler/queue/GroupFactory.scala | 5 +++
.../linkis/scheduler/queue/LoopArrayQueue.scala | 7 +++-
.../queue/fifoqueue/FIFOUserConsumer.scala | 45 +++++++++++++++++++---
4 files changed, 50 insertions(+), 8 deletions(-)
diff --git a/linkis-commons/linkis-scheduler/src/main/scala/org/apache/linkis/scheduler/queue/Consumer.scala b/linkis-commons/linkis-scheduler/src/main/scala/org/apache/linkis/scheduler/queue/Consumer.scala
index 50dce2ca1..165a27436 100644
--- a/linkis-commons/linkis-scheduler/src/main/scala/org/apache/linkis/scheduler/queue/Consumer.scala
+++ b/linkis-commons/linkis-scheduler/src/main/scala/org/apache/linkis/scheduler/queue/Consumer.scala
@@ -41,7 +41,6 @@ abstract class Consumer(schedulerContext: SchedulerContext, executeService: Exec
def start(): Unit
def shutdown(): Unit = {
- logger.info(s"$toString is ready to stop!")
terminate = true
logger.info(s"$toString stopped!")
}
diff --git a/linkis-commons/linkis-scheduler/src/main/scala/org/apache/linkis/scheduler/queue/GroupFactory.scala b/linkis-commons/linkis-scheduler/src/main/scala/org/apache/linkis/scheduler/queue/GroupFactory.scala
index f3471b07d..be1716f23 100644
--- a/linkis-commons/linkis-scheduler/src/main/scala/org/apache/linkis/scheduler/queue/GroupFactory.scala
+++ b/linkis-commons/linkis-scheduler/src/main/scala/org/apache/linkis/scheduler/queue/GroupFactory.scala
@@ -19,6 +19,11 @@ package org.apache.linkis.scheduler.queue
abstract class GroupFactory {
+ /**
+ * Create a Group and set the concurrency limit of the group
+ * @param event
+ * @return
+ */
def getOrCreateGroup(event: SchedulerEvent): Group
def getGroup(groupName: String): Group
diff --git a/linkis-commons/linkis-scheduler/src/main/scala/org/apache/linkis/scheduler/queue/LoopArrayQueue.scala b/linkis-commons/linkis-scheduler/src/main/scala/org/apache/linkis/scheduler/queue/LoopArrayQueue.scala
index b0bbfd3c2..8bea7e52b 100644
--- a/linkis-commons/linkis-scheduler/src/main/scala/org/apache/linkis/scheduler/queue/LoopArrayQueue.scala
+++ b/linkis-commons/linkis-scheduler/src/main/scala/org/apache/linkis/scheduler/queue/LoopArrayQueue.scala
@@ -40,7 +40,12 @@ class LoopArrayQueue(var group: Group) extends ConsumeQueue with Logging {
override def getWaitingEvents: Array[SchedulerEvent] = {
eventQueue synchronized {
- toIndexedSeq.filter(x => x.getState.equals(SchedulerEventState.Inited)).toArray
+ toIndexedSeq
+ .filter(x =>
+ x.getState.equals(SchedulerEventState.Inited) || x.getState
+ .equals(SchedulerEventState.Scheduled)
+ )
+ .toArray
}
}
diff --git a/linkis-commons/linkis-scheduler/src/main/scala/org/apache/linkis/scheduler/queue/fifoqueue/FIFOUserConsumer.scala b/linkis-commons/linkis-scheduler/src/main/scala/org/apache/linkis/scheduler/queue/fifoqueue/FIFOUserConsumer.scala
index 2a40c2517..d541d8a2e 100644
--- a/linkis-commons/linkis-scheduler/src/main/scala/org/apache/linkis/scheduler/queue/fifoqueue/FIFOUserConsumer.scala
+++ b/linkis-commons/linkis-scheduler/src/main/scala/org/apache/linkis/scheduler/queue/fifoqueue/FIFOUserConsumer.scala
@@ -73,6 +73,8 @@ class FIFOUserConsumer(
override def getRunningEvents: Array[SchedulerEvent] =
getEvents(e => e.isRunning || e.isWaitForRetry)
+ protected def getSchedulerContext: SchedulerContext = schedulerContext
+
private def getEvents(op: SchedulerEvent => Boolean): Array[SchedulerEvent] = {
val result = ArrayBuffer[SchedulerEvent]()
runningJobs.filter(_ != null).filter(x => op(x)).foreach(result += _)
@@ -82,16 +84,28 @@ class FIFOUserConsumer(
override def run(): Unit = {
Thread.currentThread().setName(s"${toString}Thread")
logger.info(s"$toString thread started!")
- while (!terminate) {
- Utils.tryAndError(loop())
- Utils.tryAndError(Thread.sleep(10))
+ while (!terminate) Utils.tryAndError {
+ loop()
+ Thread.sleep(10)
}
logger.info(s"$toString thread stopped!")
}
protected def askExecutorGap(): Unit = {}
+ /**
+ * Task scheduling interception is used to judge the rules of task operation, and to judge other
+ * task rules based on Group. For example, Entrance makes Creator-level task judgment.
+ */
+ protected def runScheduleIntercept(): Boolean = {
+ true
+ }
+
protected def loop(): Unit = {
+ if (!runScheduleIntercept()) {
+ Utils.tryQuietly(Thread.sleep(1000))
+ return
+ }
var isRetryJob = false
def getWaitForRetryEvent: Option[SchedulerEvent] = {
val waitForRetryJobs = runningJobs.filter(job => job != null && job.isJobCanRetry)
@@ -110,7 +124,7 @@ class FIFOUserConsumer(
if (event.isEmpty) {
val completedNums = runningJobs.filter(job => job == null || job.isCompleted)
if (completedNums.length < 1) {
- Utils.tryQuietly(Thread.sleep(1000)) // TODO 还可以优化，通过实现JobListener进行优化
+ Utils.tryQuietly(Thread.sleep(1000))
return
}
while (event.isEmpty) {
@@ -119,7 +133,12 @@ class FIFOUserConsumer(
if (
takeEvent.exists(e =>
Utils.tryCatch(e.turnToScheduled()) { t =>
- takeEvent.get.asInstanceOf[Job].onFailure("Job状态翻转为Scheduled失败！", t)
+ takeEvent.get
+ .asInstanceOf[Job]
+ .onFailure(
+ "Failed to change the job status to Scheduled(Job状态翻转为Scheduled失败)",
+ t
+ )
false
}
)
@@ -174,7 +193,7 @@ class FIFOUserConsumer(
)
)
case error: Throwable =>
- job.onFailure("请求引擎失败，可能是由于后台进程错误!请联系管理员", error)
+ job.onFailure("Failed to request EngineConn", error)
if (job.isWaitForRetry) {
logger.warn(s"Ask executor for Job $job failed, wait for the next retry!", error)
if (!isRetryJob) putToRunningJobs(job)
@@ -190,6 +209,20 @@ class FIFOUserConsumer(
override def shutdown(): Unit = {
future.cancel(true)
+ val waitEvents = queue.getWaitingEvents
+ if (waitEvents.nonEmpty) {
+ waitEvents.foreach {
+ case job: Job =>
+ job.onFailure("Your job will be marked as canceled because the consumer be killed", null)
+ case _ =>
+ }
+ }
+
+ this.runningJobs.foreach { job =>
+ if (job != null && !job.isCompleted) {
+ job.onFailure("Your job will be marked as canceled because the consumer be killed", null)
+ }
+ }
super.shutdown()
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@linkis.apache.org
For additional commands, e-mail: commits-help@linkis.apache.org
[linkis] 05/11: add ps datasource server conf
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
peacewong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/linkis.git
commit 01ae62a563f838aaa7d58977814da7472d3364d9
Author: peacewong <wp...@gmail.com>
AuthorDate: Fri Sep 22 14:56:19 2023 +0800
add ps datasource server conf
---
linkis-commons/linkis-mybatis/pom.xml | 26 ++++++++++++++++++++++
.../apache/linkis/rpc/conf/RPCConfiguration.scala | 10 +++++++++
.../linkis/rpc/conf/RPCConfigurationTest.scala | 2 --
3 files changed, 36 insertions(+), 2 deletions(-)
diff --git a/linkis-commons/linkis-mybatis/pom.xml b/linkis-commons/linkis-mybatis/pom.xml
index b87a1ed66..1481f2008 100644
--- a/linkis-commons/linkis-mybatis/pom.xml
+++ b/linkis-commons/linkis-mybatis/pom.xml
@@ -37,6 +37,32 @@
<dependency>
<groupId>com.baomidou</groupId>
<artifactId>mybatis-plus-boot-starter</artifactId>
+ <exclusions>
+ <exclusion>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-autoconfigure</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.springframework</groupId>
+ <artifactId>spring-beans</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.springframework</groupId>
+ <artifactId>spring-jdbc</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.zaxxer</groupId>
+ <artifactId>HikariCP</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.springframework</groupId>
+ <artifactId>spring-jdbc</artifactId>
</dependency>
<dependency>
<groupId>com.github.pagehelper</groupId>
diff --git a/linkis-commons/linkis-rpc/src/main/scala/org/apache/linkis/rpc/conf/RPCConfiguration.scala b/linkis-commons/linkis-rpc/src/main/scala/org/apache/linkis/rpc/conf/RPCConfiguration.scala
index 32f12da27..3a34e42a0 100644
--- a/linkis-commons/linkis-rpc/src/main/scala/org/apache/linkis/rpc/conf/RPCConfiguration.scala
+++ b/linkis-commons/linkis-rpc/src/main/scala/org/apache/linkis/rpc/conf/RPCConfiguration.scala
@@ -80,6 +80,16 @@ object RPCConfiguration {
CommonVars("wds.linkis.gateway.conf.linkismanager.list", "linkisManager,engineplugin").getValue
.split(",")
+ val LINKIS_DATASOURCE_SERVICE_NAME: CommonVars[String] =
+ CommonVars("linkis.gateway.conf.linkisdatasource.name", "linkis-ps-datasource")
+
+ val LINKIS_DATASOURCE_SERVICE_LIST: Array[String] =
+ CommonVars(
+ "linkis.gateway.conf.linkisdatasource.list",
+ "data-source-manager,metadataquery,datasource,metadataQuery,metadatamanager"
+ ).getValue
+ .split(",")
+
val BDP_RPC_INSTANCE_ALIAS_SERVICE_REFRESH_INTERVAL: CommonVars[TimeType] =
CommonVars("wds.linkis.rpc.instancealias.refresh.interval", new TimeType("3s"))
diff --git a/linkis-commons/linkis-rpc/src/test/scala/org/apache/linkis/rpc/conf/RPCConfigurationTest.scala b/linkis-commons/linkis-rpc/src/test/scala/org/apache/linkis/rpc/conf/RPCConfigurationTest.scala
index 5aaadd133..ed1d39ce8 100644
--- a/linkis-commons/linkis-rpc/src/test/scala/org/apache/linkis/rpc/conf/RPCConfigurationTest.scala
+++ b/linkis-commons/linkis-rpc/src/test/scala/org/apache/linkis/rpc/conf/RPCConfigurationTest.scala
@@ -37,7 +37,6 @@ class RPCConfigurationTest {
val enablepublicservice = RPCConfiguration.ENABLE_PUBLIC_SERVICE.getValue
val publicserviceapplicationname = RPCConfiguration.PUBLIC_SERVICE_APPLICATION_NAME.getValue
val publicservicelist = RPCConfiguration.PUBLIC_SERVICE_LIST
- val metadataqueryservicelist = RPCConfiguration.METADATAQUERY_SERVICE_LIST
Assertions.assertTrue(25 == bdprpcbroadcastthreadsize.intValue())
Assertions.assertTrue(400 == bdprpcreceiverasynconsumerthreadmax.intValue())
@@ -48,7 +47,6 @@ class RPCConfigurationTest {
Assertions.assertTrue(enablepublicservice)
Assertions.assertEquals("linkis-ps-publicservice", publicserviceapplicationname)
Assertions.assertTrue(publicservicelist.size > 0)
- Assertions.assertTrue(metadataqueryservicelist.size > 0)
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@linkis.apache.org
For additional commands, e-mail: commits-help@linkis.apache.org
[linkis] 06/11: optimize storage module
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
peacewong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/linkis.git
commit dc8a7ad3c32c032ddbba0dacd48e9f5cdaaff6e0
Author: peacewong <wp...@gmail.com>
AuthorDate: Sun Oct 1 23:55:08 2023 +0800
optimize storage module
---
.../java/org/apache/linkis/common/io/FsPath.java | 1 +
.../linkis/storage/conf/LinkisStorageConf.java | 2 +-
.../linkis/storage/csv/StorageCSVWriter.java | 15 +++++++-----
.../org/apache/linkis/storage/domain/DataType.java | 15 ++++++++++--
.../org/apache/linkis/storage/domain/Dolphin.java | 11 +++++++++
.../linkis/storage/exception/StorageErrorCode.java | 5 +++-
.../storage/resultset/ResultSetReaderFactory.java | 1 +
.../storage/resultset/StorageResultSetReader.java | 27 +++++++++-------------
.../storage/resultset/table/TableRecord.java | 9 --------
.../resultset/table/TableResultDeserializer.java | 6 ++---
.../resultset/table/TableResultSerializer.java | 2 +-
.../apache/linkis/storage/source/FileSource.java | 1 +
.../linkis/storage/source/ResultsetFileSource.java | 14 +++++++++--
13 files changed, 68 insertions(+), 41 deletions(-)
diff --git a/linkis-commons/linkis-common/src/main/java/org/apache/linkis/common/io/FsPath.java b/linkis-commons/linkis-common/src/main/java/org/apache/linkis/common/io/FsPath.java
index b5e9af86f..a1c0839b2 100644
--- a/linkis-commons/linkis-common/src/main/java/org/apache/linkis/common/io/FsPath.java
+++ b/linkis-commons/linkis-common/src/main/java/org/apache/linkis/common/io/FsPath.java
@@ -24,6 +24,7 @@ import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.file.FileSystems;
+import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.attribute.PosixFilePermission;
import java.nio.file.attribute.PosixFilePermissions;
diff --git a/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/conf/LinkisStorageConf.java b/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/conf/LinkisStorageConf.java
index 66fafeacd..74950c15f 100644
--- a/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/conf/LinkisStorageConf.java
+++ b/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/conf/LinkisStorageConf.java
@@ -39,7 +39,7 @@ public class LinkisStorageConf {
public static final String FILE_TYPE =
CommonVars.apply(
"wds.linkis.storage.file.type",
- "dolphin,sql,scala,py,hql,python,out,log,text,sh,jdbc,ngql,psql,fql,tsql")
+ "dolphin,sql,scala,py,hql,python,out,log,text,txt,sh,jdbc,ngql,psql,fql,tsql")
.getValue();
private static volatile String[] fileTypeArr = null;
diff --git a/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/csv/StorageCSVWriter.java b/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/csv/StorageCSVWriter.java
index ba4d877f3..d98be4033 100644
--- a/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/csv/StorageCSVWriter.java
+++ b/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/csv/StorageCSVWriter.java
@@ -91,14 +91,17 @@ public class StorageCSVWriter extends CSVFsWriter {
private String compact(String[] row) {
String quotationMarks = "\"";
+ String dealNewlineSymbolMarks = "\n";
StringBuilder rowBuilder = new StringBuilder();
for (String value : row) {
- String decoratedValue =
- StringUtils.isBlank(value)
- ? value
- : quoteRetouchEnable
- ? quotationMarks + value.replaceAll(quotationMarks, "") + quotationMarks
- : value;
+ String decoratedValue = value;
+ if (StringUtils.isNotBlank(value)) {
+ if (quoteRetouchEnable) {
+ decoratedValue = quotationMarks + value.replaceAll(quotationMarks, "") + quotationMarks;
+ }
+ decoratedValue = decoratedValue.replaceAll(dealNewlineSymbolMarks, " ");
+ logger.debug("decorateValue with input: {} output: {} ", value, decoratedValue);
+ }
rowBuilder.append(decoratedValue).append(delimiter);
}
if (rowBuilder.length() > 0 && rowBuilder.toString().endsWith(delimiter)) {
diff --git a/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/domain/DataType.java b/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/domain/DataType.java
index ad9e0ee88..6808f693e 100644
--- a/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/domain/DataType.java
+++ b/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/domain/DataType.java
@@ -26,6 +26,8 @@ import java.util.regex.Pattern;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import static org.apache.linkis.storage.domain.Dolphin.LINKIS_NULL;
+
public enum DataType {
NullType("void", 0),
StringType("string", 12),
@@ -63,7 +65,8 @@ public enum DataType {
public static final String LOWCASE_NULL_VALUE = "null";
// TODO Change to fine-grained regular expressions(改为精细化正则表达式)
- public static final Pattern DECIMAL_REGEX = Pattern.compile("^decimal\\(\\d*\\,\\d*\\)");
+ public static final Pattern DECIMAL_REGEX =
+ Pattern.compile("^decimal\\(\\s*\\d*\\s*,\\s*\\d*\\s*\\)");
public static final Pattern SHORT_REGEX = Pattern.compile("^short.*");
public static final Pattern INT_REGEX = Pattern.compile("^int.*");
@@ -130,7 +133,11 @@ public enum DataType {
}
public static Object toValue(DataType dataType, String value) {
+
Object result = null;
+ if (isLinkisNull(value)) {
+ return result;
+ }
try {
switch (dataType) {
case NullType:
@@ -187,12 +194,16 @@ public enum DataType {
result = value;
}
} catch (Exception e) {
- logger.debug("Failed to " + value + " switch to dataType:", e);
+ logger.debug("Failed to {} switch to dataType:", value, e);
result = value;
}
return result;
}
+ public static boolean isLinkisNull(String value) {
+ return value == null || value.equals(LINKIS_NULL);
+ }
+
public static boolean isNull(String value) {
return value == null || value.equals(NULL_VALUE) || value.trim().equals("");
}
diff --git a/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/domain/Dolphin.java b/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/domain/Dolphin.java
index b6badd284..35c71295e 100644
--- a/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/domain/Dolphin.java
+++ b/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/domain/Dolphin.java
@@ -59,6 +59,9 @@ public class Dolphin {
public static final String NULL = "NULL";
public static final byte[] NULL_BYTES = "NULL".getBytes(Charset.forName("utf-8"));
+ public static final String LINKIS_NULL = "LINKIS_NULL";
+ public static final byte[] LINKIS_NULL_BYTES = LINKIS_NULL.getBytes(Charset.forName("utf-8"));
+
public static final int INT_LEN = 10;
public static final int FILE_EMPTY = 31;
@@ -79,6 +82,14 @@ public class Dolphin {
return new String(bytes, start, len, Dolphin.CHAR_SET);
}
+ public static String toStringValue(String value) {
+ if (LINKIS_NULL.equals(value)) {
+ return NULL;
+ } else {
+ return value;
+ }
+ }
+
/**
* Read an integer value that converts the array to a byte of length 10 bytes
* 读取整数值，该值为将数组转换为10字节长度的byte
diff --git a/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/exception/StorageErrorCode.java b/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/exception/StorageErrorCode.java
index 3c82ceb52..fad0d83a1 100644
--- a/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/exception/StorageErrorCode.java
+++ b/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/exception/StorageErrorCode.java
@@ -20,7 +20,10 @@ package org.apache.linkis.storage.exception;
public enum StorageErrorCode {
/** */
- FS_NOT_INIT(53001, "please init first(请先初始化)");
+ FS_NOT_INIT(53001, "please init first"),
+
+ INCONSISTENT_DATA(53001, "Inconsistent row data read,read %s,need rowLen %s"),
+ FS_OOM(53002, "OOM occurred while reading the file");
StorageErrorCode(int errorCode, String message) {
this.code = errorCode;
diff --git a/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/resultset/ResultSetReaderFactory.java b/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/resultset/ResultSetReaderFactory.java
index 749ec9a24..3047b715a 100644
--- a/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/resultset/ResultSetReaderFactory.java
+++ b/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/resultset/ResultSetReaderFactory.java
@@ -99,6 +99,7 @@ public class ResultSetReaderFactory {
}
Fs fs = FSFactory.getFs(resPath);
+ logger.info("Try to init Fs with path:{}", resPath.getPath());
try {
fs.init(null);
InputStream read = fs.read(resPath);
diff --git a/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/resultset/StorageResultSetReader.java b/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/resultset/StorageResultSetReader.java
index 35f7483c8..23d65e061 100644
--- a/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/resultset/StorageResultSetReader.java
+++ b/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/resultset/StorageResultSetReader.java
@@ -33,7 +33,6 @@ import org.apache.commons.lang3.StringUtils;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
-import java.util.Arrays;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -53,7 +52,6 @@ public class StorageResultSetReader<K extends MetaData, V extends Record>
private Fs fs;
private final int READ_CACHE = 1024;
- private final byte[] bytes = new byte[READ_CACHE];
public StorageResultSetReader(ResultSet<K, V> resultSet, InputStream inputStream) {
super(resultSet, inputStream);
@@ -84,21 +82,18 @@ public class StorageResultSetReader<K extends MetaData, V extends Record>
return null;
}
- byte[] rowBuffer = new byte[0];
int len = 0;
-
- while (rowLen > 0 && len >= 0) {
- if (rowLen > READ_CACHE) {
- len = StorageUtils.readBytes(inputStream, bytes, READ_CACHE);
- } else {
- len = StorageUtils.readBytes(inputStream, bytes, rowLen);
- }
-
- if (len > 0) {
- rowLen -= len;
- rowBuffer = Arrays.copyOf(rowBuffer, rowBuffer.length + len);
- System.arraycopy(bytes, 0, rowBuffer, rowBuffer.length - len, len);
- }
+ byte[] rowBuffer = null;
+ try {
+ rowBuffer = new byte[rowLen];
+ len = StorageUtils.readBytes(inputStream, rowBuffer, READ_CACHE);
+ } catch (OutOfMemoryError error) {
+ logger.error("Result set read oom, read size {} Byte", rowLen);
+ throw new RuntimeException(error);
+ }
+ if (len != rowLen) {
+ throw new RuntimeException(
+ "Can't get the value of the field, maybe the IO stream has been read or has been closed!(拿不到字段的值，也许IO流已读取完毕或已被关闭！)");
}
rowCount++;
return rowBuffer;
diff --git a/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/resultset/table/TableRecord.java b/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/resultset/table/TableRecord.java
index f13d42b0b..c8270714b 100644
--- a/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/resultset/table/TableRecord.java
+++ b/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/resultset/table/TableRecord.java
@@ -19,9 +19,6 @@ package org.apache.linkis.storage.resultset.table;
import org.apache.linkis.common.io.Record;
import org.apache.linkis.storage.resultset.ResultRecord;
-import org.apache.linkis.storage.utils.StorageUtils;
-
-import java.util.Arrays;
public class TableRecord implements ResultRecord {
@@ -35,10 +32,4 @@ public class TableRecord implements ResultRecord {
public Record cloneRecord() {
return new TableRecord(row.clone());
}
-
- public String[] tableRecordToString(String nullValue) {
- return Arrays.stream(row)
- .map(col -> StorageUtils.colToString(col, nullValue))
- .toArray(String[]::new);
- }
}
diff --git a/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/resultset/table/TableResultDeserializer.java b/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/resultset/table/TableResultDeserializer.java
index 7b7838fd2..7e1d6c35f 100644
--- a/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/resultset/table/TableResultDeserializer.java
+++ b/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/resultset/table/TableResultDeserializer.java
@@ -52,13 +52,13 @@ public class TableResultDeserializer extends ResultDeserializer<TableMetaData, T
List<Column> columns = new ArrayList<>();
for (int i = 0; i < colArray.length; i += 3) {
int len = Integer.parseInt(colArray[i]);
- String colName = Dolphin.getString(bytes, index, len);
+ String colName = Dolphin.toStringValue(Dolphin.getString(bytes, index, len));
index += len;
len = Integer.parseInt(colArray[i + 1]);
- String colType = Dolphin.getString(bytes, index, len);
+ String colType = Dolphin.toStringValue(Dolphin.getString(bytes, index, len));
index += len;
len = Integer.parseInt(colArray[i + 2]);
- String colComment = Dolphin.getString(bytes, index, len);
+ String colComment = Dolphin.toStringValue(Dolphin.getString(bytes, index, len));
index += len;
columns.add(new Column(colName, DataType.toDataType(colType), colComment));
}
diff --git a/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/resultset/table/TableResultSerializer.java b/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/resultset/table/TableResultSerializer.java
index 6abe4c56d..5f40aa33f 100644
--- a/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/resultset/table/TableResultSerializer.java
+++ b/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/resultset/table/TableResultSerializer.java
@@ -61,7 +61,7 @@ public class TableResultSerializer extends ResultSerializer {
int colByteLen = 0;
int length = 0;
for (Object data : line) {
- byte[] bytes = data == null ? Dolphin.NULL_BYTES : Dolphin.getBytes(data);
+ byte[] bytes = data == null ? Dolphin.LINKIS_NULL_BYTES : Dolphin.getBytes(data);
dataBytes.add(bytes);
byte[] colBytes = Dolphin.getBytes(bytes.length);
colIndex.add(colBytes);
diff --git a/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/source/FileSource.java b/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/source/FileSource.java
index cee72dfcd..0ed650186 100644
--- a/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/source/FileSource.java
+++ b/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/source/FileSource.java
@@ -135,6 +135,7 @@ public interface FileSource extends Closeable {
}
static FileSplit createResultSetFileSplit(FsPath fsPath, InputStream is) {
+ logger.info("try create result set file split with path:{}", fsPath.getPath());
ResultSet resultset = ResultSetFactory.getInstance().getResultSetByPath(fsPath);
ResultSetReader resultsetReader = ResultSetReaderFactory.getResultSetReader(resultset, is);
return new FileSplit(resultsetReader, resultset.resultSetType());
diff --git a/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/source/ResultsetFileSource.java b/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/source/ResultsetFileSource.java
index d8562b7c5..fb064a8f4 100644
--- a/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/source/ResultsetFileSource.java
+++ b/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/source/ResultsetFileSource.java
@@ -17,6 +17,7 @@
package org.apache.linkis.storage.source;
+import org.apache.linkis.storage.domain.Dolphin;
import org.apache.linkis.storage.resultset.table.TableRecord;
import org.apache.linkis.storage.utils.StorageUtils;
@@ -36,9 +37,18 @@ public class ResultsetFileSource extends AbstractFileSource {
.map(
r -> {
if (r == null || r.equals("NULL")) {
- return nullValue;
+ if (nullValue.equals(Dolphin.LINKIS_NULL)) {
+ return r;
+ } else {
+ return nullValue;
+ }
} else if (r.equals("")) {
- return getParams().getOrDefault("nullValue", "");
+ String emptyValue = getParams().getOrDefault("nullValue", "");
+ if (emptyValue.equals(Dolphin.LINKIS_NULL)) {
+ return "";
+ } else {
+ return nullValue;
+ }
} else if (r instanceof Double) {
return StorageUtils.doubleToString((Double) r);
} else {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@linkis.apache.org
For additional commands, e-mail: commits-help@linkis.apache.org
[linkis] 02/11: add default retry for 401
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
peacewong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/linkis.git
commit 2b6d80240a065c67085d36ae915cec5b5bfeaae1
Author: peacewong <wp...@gmail.com>
AuthorDate: Fri Sep 22 14:54:24 2023 +0800
add default retry for 401
---
.../linkis/httpclient/AbstractHttpClient.scala | 26 ++++++++++++++--------
.../httpclient/config/ClientConfigBuilder.scala | 9 ++++----
2 files changed, 21 insertions(+), 14 deletions(-)
diff --git a/linkis-commons/linkis-httpclient/src/main/scala/org/apache/linkis/httpclient/AbstractHttpClient.scala b/linkis-commons/linkis-httpclient/src/main/scala/org/apache/linkis/httpclient/AbstractHttpClient.scala
index d901bb902..ec61ebd66 100644
--- a/linkis-commons/linkis-httpclient/src/main/scala/org/apache/linkis/httpclient/AbstractHttpClient.scala
+++ b/linkis-commons/linkis-httpclient/src/main/scala/org/apache/linkis/httpclient/AbstractHttpClient.scala
@@ -143,23 +143,31 @@ abstract class AbstractHttpClient(clientConfig: ClientConfig, clientName: String
val req = prepareReq(action)
val startTime = System.currentTimeMillis
val response = executeRequest(req, Some(waitTime).filter(_ > 0))
+ val taken = System.currentTimeMillis - startTime
+ attempts.add(taken)
+ val costTime = ByteTimeUtils.msDurationToString(taken)
+ logger.info(
+ s"invoke ${req.getURI} get status ${response.getStatusLine.getStatusCode} taken: ${costTime}."
+ )
if (response.getStatusLine.getStatusCode == 401) {
tryLogin(action, getRequestUrl(action), true)
- logger.info("The user is not logged in, please log in first, you can set a retry")
val msg = Utils.tryCatch(EntityUtils.toString(response.getEntity)) { t =>
logger.warn("failed to parse entity", t)
""
}
IOUtils.closeQuietly(response)
- throw new HttpClientRetryException(
- "The user is not logged in, please log in first, you can set a retry, message: " + msg
- )
+ if (attempts.size() <= 1) {
+ logger.info("The user is not logged in, default retry once")
+ addAttempt()
+ } else {
+ logger.info("The user is not logged in, you can set a retry")
+ throw new HttpClientRetryException(
+ "The user is not logged in, please log in first, you can set a retry, message: " + msg
+ )
+ }
+ } else {
+ response
}
- val taken = System.currentTimeMillis - startTime
- attempts.add(taken)
- val costTime = ByteTimeUtils.msDurationToString(taken)
- logger.info(s"invoke ${req.getURI} taken: ${costTime}.")
- response
}
val response =
diff --git a/linkis-commons/linkis-httpclient/src/main/scala/org/apache/linkis/httpclient/config/ClientConfigBuilder.scala b/linkis-commons/linkis-httpclient/src/main/scala/org/apache/linkis/httpclient/config/ClientConfigBuilder.scala
index 26170a457..b1fc579f3 100644
--- a/linkis-commons/linkis-httpclient/src/main/scala/org/apache/linkis/httpclient/config/ClientConfigBuilder.scala
+++ b/linkis-commons/linkis-httpclient/src/main/scala/org/apache/linkis/httpclient/config/ClientConfigBuilder.scala
@@ -17,9 +17,9 @@
package org.apache.linkis.httpclient.config
+import org.apache.linkis.common.exception.LinkisRetryException
import org.apache.linkis.common.utils.{DefaultRetryHandler, RetryHandler}
import org.apache.linkis.httpclient.authentication.AuthenticationStrategy
-import org.apache.linkis.httpclient.exception.HttpClientRetryException
import org.apache.linkis.httpclient.loadbalancer.LoadBalancerStrategy
import scala.concurrent.duration.TimeUnit
@@ -39,11 +39,10 @@ class ClientConfigBuilder protected () {
protected var readTimeout: Long = _
protected var maxConnection: Int = _
protected var retryEnabled: Boolean = true
- protected var retryHandler: RetryHandler = buildDefaultRetryHandler()
- def buildDefaultRetryHandler(): RetryHandler = {
- retryHandler = new DefaultRetryHandler
- retryHandler.addRetryException(classOf[HttpClientRetryException])
+ protected var retryHandler: RetryHandler = {
+ val retryHandler = new DefaultRetryHandler
+ retryHandler.addRetryException(classOf[LinkisRetryException])
retryHandler
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@linkis.apache.org
For additional commands, e-mail: commits-help@linkis.apache.org
[linkis] 01/11: 1. add result set sot utils class 2. optimize variable operation logic
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
peacewong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/linkis.git
commit fcd6c7c0b554b855fcbad68a0113f422a1982e8a
Author: peacewong <wp...@gmail.com>
AuthorDate: Fri Sep 22 12:09:28 2023 +0800
1. add result set sot utils class
2. optimize variable operation logic
---
.../java/org/apache/linkis/common/io/FsPath.java | 5 +
.../apache/linkis/common/utils/ByteTimeUtils.java | 13 +-
.../apache/linkis/common/utils/ResultSetUtils.java | 57 ++++++++
.../common/utils/VariableOperationUtils.java | 43 ++++--
.../apache/linkis/common/utils/VariableUtils.scala | 24 +++-
.../linkis/common/utils/ByteTimeUtilsTest.java | 152 +++++++++++++++++++--
.../linkis/common/utils/VariableUtilsTest.scala | 17 +++
7 files changed, 284 insertions(+), 27 deletions(-)
diff --git a/linkis-commons/linkis-common/src/main/java/org/apache/linkis/common/io/FsPath.java b/linkis-commons/linkis-common/src/main/java/org/apache/linkis/common/io/FsPath.java
index 7908e5a50..b5e9af86f 100644
--- a/linkis-commons/linkis-common/src/main/java/org/apache/linkis/common/io/FsPath.java
+++ b/linkis-commons/linkis-common/src/main/java/org/apache/linkis/common/io/FsPath.java
@@ -23,6 +23,7 @@ import java.io.File;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
+import java.nio.file.FileSystems;
import java.nio.file.Paths;
import java.nio.file.attribute.PosixFilePermission;
import java.nio.file.attribute.PosixFilePermissions;
@@ -168,6 +169,10 @@ public class FsPath {
return new File(uri);
}
+ public Path toPath() {
+ return FileSystems.getDefault().getPath(uri.toString());
+ }
+
public String getUriString() {
return uri.toString();
}
diff --git a/linkis-commons/linkis-common/src/main/java/org/apache/linkis/common/utils/ByteTimeUtils.java b/linkis-commons/linkis-common/src/main/java/org/apache/linkis/common/utils/ByteTimeUtils.java
index d23f4a086..0ecb3dc2a 100644
--- a/linkis-commons/linkis-common/src/main/java/org/apache/linkis/common/utils/ByteTimeUtils.java
+++ b/linkis-commons/linkis-common/src/main/java/org/apache/linkis/common/utils/ByteTimeUtils.java
@@ -213,7 +213,6 @@ public class ByteTimeUtils {
} else {
throw new NumberFormatException("Failed to parse byte string: " + str);
}
- suffix = suffix.toLowerCase();
// Check for invalid suffixes
if (suffix != null && !byteSuffixes.containsKey(suffix)) {
throw new NumberFormatException("Invalid suffix: \"" + suffix + "\"");
@@ -297,6 +296,18 @@ public class ByteTimeUtils {
return parseByteString(str, ByteUnit.GiB);
}
+ /**
+ * Convert a passed byte string (e.g. -50b, -100k, or -250m) to gibibytes for internal use.
+ *
+ * <p>If no suffix is provided, the passed number is assumed to be in gibibytes.
+ */
+ public static long negativeByteStringAsGb(String str) {
+ if (str.startsWith("-")) {
+ return Math.negateExact(parseByteString(str.substring(1), ByteUnit.GiB));
+ }
+ return parseByteString(str, ByteUnit.GiB);
+ }
+
/**
* Returns a byte array with the buffer's contents, trying to avoid copying the data if possible.
*/
diff --git a/linkis-commons/linkis-common/src/main/java/org/apache/linkis/common/utils/ResultSetUtils.java b/linkis-commons/linkis-common/src/main/java/org/apache/linkis/common/utils/ResultSetUtils.java
new file mode 100644
index 000000000..a367b38b8
--- /dev/null
+++ b/linkis-commons/linkis-common/src/main/java/org/apache/linkis/common/utils/ResultSetUtils.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.linkis.common.utils;
+
+import org.apache.linkis.common.io.FsPath;
+
+import java.io.File;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+public class ResultSetUtils {
+
+ // Sort in ASC order by numx in the result set _numx.dolphin file name
+ public static Comparator<FsPath> getResultSetFileComparatorOrderByNameNum() {
+
+ Comparator<FsPath> comparator =
+ (o1, o2) -> {
+ // get the num of file name
+ String regx = "\\d+";
+
+ String[] res1 = o1.getPath().split(File.separator);
+ String fileName1 = res1[res1.length - 1];
+ Matcher matcher1 = Pattern.compile(regx).matcher(fileName1);
+ int num1 = matcher1.find() ? Integer.parseInt(matcher1.group()) : Integer.MAX_VALUE;
+
+ String[] res2 = o2.getPath().split(File.separator);
+ String fileName2 = res2[res2.length - 1];
+ Matcher matcher2 = Pattern.compile(regx).matcher(fileName2);
+ int num2 = matcher2.find() ? Integer.parseInt(matcher2.group()) : Integer.MAX_VALUE;
+
+ return num1 - num2;
+ };
+ return comparator;
+ }
+
+ public static void sortByNameNum(List<FsPath> fsPathList) {
+ Collections.sort(fsPathList, getResultSetFileComparatorOrderByNameNum());
+ }
+}
diff --git a/linkis-commons/linkis-common/src/main/java/org/apache/linkis/common/utils/VariableOperationUtils.java b/linkis-commons/linkis-common/src/main/java/org/apache/linkis/common/utils/VariableOperationUtils.java
index b891f99d1..615472474 100644
--- a/linkis-commons/linkis-common/src/main/java/org/apache/linkis/common/utils/VariableOperationUtils.java
+++ b/linkis-commons/linkis-common/src/main/java/org/apache/linkis/common/utils/VariableOperationUtils.java
@@ -28,7 +28,10 @@ import java.util.Date;
import java.util.Iterator;
import java.util.Map;
+import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.json.JsonMapper;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
@@ -49,6 +52,9 @@ public class VariableOperationUtils {
private static final String[] CYCLES =
new String[] {CYCLE_YEAR, CYCLE_MONTH, CYCLE_DAY, CYCLE_HOUR, CYCLE_MINUTE, CYCLE_SECOND};
+ private static final ObjectMapper mapper =
+ JsonMapper.builder().enable(DeserializationFeature.FAIL_ON_TRAILING_TOKENS).build();
+
/**
* yyyy-MM-dd HH:mm:ss
*
@@ -78,30 +84,44 @@ public class VariableOperationUtils {
* @param str
* @return
*/
+ @Deprecated
public static String replaces(ZonedDateTime dateTime, String str)
throws VariableOperationFailedException {
- return replaces(dateTime, str, true);
+ try {
+ JsonNode rootNode = mapper.readTree(str);
+ if (rootNode.isArray() || rootNode.isObject()) {
+ replaceJson(dateTime, rootNode);
+ return rootNode.toString();
+ }
+ } catch (Exception e) {
+ return replace(dateTime, str);
+ }
+ return replace(dateTime, str);
}
/**
* json support variable operation
*
+ * @param codeType
* @param dateTime
* @param str
- * @param format
* @return
*/
- public static String replaces(ZonedDateTime dateTime, String str, boolean format)
+ public static String replaces(String codeType, ZonedDateTime dateTime, String str)
throws VariableOperationFailedException {
- try {
- JsonNode rootNode = JsonUtils.jackson().readTree(str);
- if (rootNode.isArray() || rootNode.isObject()) {
- replaceJson(dateTime, rootNode);
- return rootNode.toString();
+ String languageType = CodeAndRunTypeUtils.getLanguageTypeByCodeType(codeType, "");
+ if (languageType.equals(CodeAndRunTypeUtils.LANGUAGE_TYPE_JSON())) {
+ try {
+ JsonNode rootNode = mapper.readTree(str);
+ if (rootNode.isArray() || rootNode.isObject()) {
+ replaceJson(dateTime, rootNode);
+ return rootNode.toString();
+ }
+ } catch (Exception e) {
+ return replace(dateTime, str);
}
- } catch (Exception e) {
- return replace(dateTime, str);
}
+
return replace(dateTime, str);
}
@@ -197,8 +217,7 @@ public class VariableOperationUtils {
} else if (temp.isObject()) {
replaceJson(dateTime, temp);
} else {
- arrayNode.remove(i);
- arrayNode.insert(i, replace(dateTime, temp.toString()));
+ arrayNode.set(i, replace(dateTime, temp.toString()));
}
}
} else if (object.isObject()) {
diff --git a/linkis-commons/linkis-common/src/main/scala/org/apache/linkis/common/utils/VariableUtils.scala b/linkis-commons/linkis-common/src/main/scala/org/apache/linkis/common/utils/VariableUtils.scala
index 30bdeb4b1..6c5bd7cf3 100644
--- a/linkis-commons/linkis-common/src/main/scala/org/apache/linkis/common/utils/VariableUtils.scala
+++ b/linkis-commons/linkis-common/src/main/scala/org/apache/linkis/common/utils/VariableUtils.scala
@@ -143,9 +143,10 @@ object VariableUtils extends Logging {
}
initAllDateVars(run_date, nameAndType)
val codeOperation = parserVar(code, nameAndType)
- parserDate(codeOperation, run_date)
+ parserDate(codeType, codeOperation, run_date)
}
+ @deprecated
private def parserDate(code: String, run_date: CustomDateType): String = {
if (Configuration.VARIABLE_OPERATION) {
val zonedDateTime: ZonedDateTime = VariableOperationUtils.toZonedDateTime(run_date.getDate)
@@ -155,6 +156,15 @@ object VariableUtils extends Logging {
}
}
+ private def parserDate(codeType: String, code: String, run_date: CustomDateType): String = {
+ if (Configuration.VARIABLE_OPERATION) {
+ val zonedDateTime: ZonedDateTime = VariableOperationUtils.toZonedDateTime(run_date.getDate)
+ VariableOperationUtils.replaces(codeType, zonedDateTime, code)
+ } else {
+ code
+ }
+ }
+
private def initAllDateVars(
run_date: CustomDateType,
nameAndType: mutable.Map[String, variable.VariableType]
@@ -337,7 +347,7 @@ object VariableUtils extends Logging {
*
* @param code
* :code
- * @param codeType
+ * @param languageType
* :SQL,PYTHON
* @return
*/
@@ -346,27 +356,37 @@ object VariableUtils extends Logging {
var varString: String = null
var errString: String = null
+ var rightVarString: String = null
languageType match {
case CodeAndRunTypeUtils.LANGUAGE_TYPE_SQL =>
varString = """\s*--@set\s*.+\s*"""
+ rightVarString = """^\s*--@set\s*.+\s*"""
errString = """\s*--@.*"""
case CodeAndRunTypeUtils.LANGUAGE_TYPE_PYTHON | CodeAndRunTypeUtils.LANGUAGE_TYPE_SHELL =>
varString = """\s*#@set\s*.+\s*"""
+ rightVarString = """^\s*#@set\s*.+\s*"""
errString = """\s*#@"""
case CodeAndRunTypeUtils.LANGUAGE_TYPE_SCALA =>
varString = """\s*//@set\s*.+\s*"""
+ rightVarString = """^\s*//@set\s*.+\s*"""
errString = """\s*//@.+"""
case CodeAndRunTypeUtils.LANGUAGE_TYPE_JAVA =>
varString = """\s*!!@set\s*.+\s*"""
+ rightVarString = """^\s*!!@set\s*.+\s*"""
case _ =>
return nameAndValue
}
val customRegex = varString.r.unanchored
+ val customRightRegex = rightVarString.r.unanchored
val errRegex = errString.r.unanchored
code.split("\n").foreach { str =>
{
+
+ if (customRightRegex.unapplySeq(str).size < customRegex.unapplySeq(str).size) {
+ logger.warn(s"code:$str is wrong custom variable format!!!")
+ }
str match {
case customRegex() =>
val clearStr = if (str.endsWith(";")) str.substring(0, str.length - 1) else str
diff --git a/linkis-commons/linkis-common/src/test/java/org/apache/linkis/common/utils/ByteTimeUtilsTest.java b/linkis-commons/linkis-common/src/test/java/org/apache/linkis/common/utils/ByteTimeUtilsTest.java
index 7b90f053c..f548d89d4 100644
--- a/linkis-commons/linkis-common/src/test/java/org/apache/linkis/common/utils/ByteTimeUtilsTest.java
+++ b/linkis-commons/linkis-common/src/test/java/org/apache/linkis/common/utils/ByteTimeUtilsTest.java
@@ -17,31 +17,159 @@
package org.apache.linkis.common.utils;
+import java.util.function.Function;
+
+import com.google.common.collect.ImmutableMap;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
-import static org.junit.jupiter.api.Assertions.*;
-
class ByteTimeUtilsTest {
+ private static final ImmutableMap<String, Function<String, Long>> opFunction =
+ ImmutableMap.<String, Function<String, Long>>builder()
+ .put("byteStringAsBytes", tar -> ByteTimeUtils.byteStringAsBytes(tar))
+ .put("byteStringAsKb", tar -> ByteTimeUtils.byteStringAsKb(tar))
+ .put("byteStringAsMb", tar -> ByteTimeUtils.byteStringAsMb(tar))
+ .put("byteStringAsGb", tar -> ByteTimeUtils.byteStringAsGb(tar))
+ .build();
+
+ private static final ImmutableMap<String, Long> convertToByte =
+ ImmutableMap.<String, Long>builder()
+ .put("1", 1l)
+ .put("1b", 1l)
+ .put("1B", 1l)
+ .put("1k", 1024l)
+ .put("1K", 1024l)
+ .put("1kb", 1024l)
+ .put("1Kb", 1024l)
+ .put("1kB", 1024l)
+ .put("1KB", 1024l)
+ .put("1m", 1024l * 1024l)
+ .put("1M", 1024l * 1024l)
+ .put("1mb", 1024l * 1024l)
+ .put("1Mb", 1024l * 1024l)
+ .put("1mB", 1024l * 1024l)
+ .put("1MB", 1024l * 1024l)
+ .put("1g", 1024l * 1024l * 1024l)
+ .put("1G", 1024l * 1024l * 1024l)
+ .put("1gb", 1024l * 1024l * 1024l)
+ .put("1gB", 1024l * 1024l * 1024l)
+ .put("1Gb", 1024l * 1024l * 1024l)
+ .put("1GB", 1024l * 1024l * 1024l)
+ .put("1t", 1024l * 1024l * 1024l * 1024l)
+ .put("1T", 1024l * 1024l * 1024l * 1024l)
+ .put("1tb", 1024l * 1024l * 1024l * 1024l)
+ .put("1Tb", 1024l * 1024l * 1024l * 1024l)
+ .put("1tB", 1024l * 1024l * 1024l * 1024l)
+ .put("1TB", 1024l * 1024l * 1024l * 1024l)
+ .put("1p", 1024l * 1024l * 1024l * 1024l * 1024l)
+ .put("1P", 1024l * 1024l * 1024l * 1024l * 1024l)
+ .put("1pb", 1024l * 1024l * 1024l * 1024l * 1024l)
+ .put("1Pb", 1024l * 1024l * 1024l * 1024l * 1024l)
+ .put("1pB", 1024l * 1024l * 1024l * 1024l * 1024l)
+ .put("1PB", 1024l * 1024l * 1024l * 1024l * 1024l)
+ .build();
+
+ private static final ImmutableMap<String, Long> convertToKB =
+ ImmutableMap.<String, Long>builder()
+ .put("1", 1l)
+ .put("1024b", 1l)
+ .put("1024B", 1l)
+ .put("1k", 1l)
+ .put("1K", 1l)
+ .put("1kb", 1l)
+ .put("1Kb", 1l)
+ .put("1kB", 1l)
+ .put("1KB", 1l)
+ .put("1m", 1024l)
+ .put("1M", 1024l)
+ .put("1mb", 1024l)
+ .put("1Mb", 1024l)
+ .put("1mB", 1024l)
+ .put("1MB", 1024l)
+ .put("1g", 1024l * 1024l)
+ .put("1G", 1024l * 1024l)
+ .put("1gb", 1024l * 1024l)
+ .put("1gB", 1024l * 1024l)
+ .put("1Gb", 1024l * 1024l)
+ .put("1GB", 1024l * 1024l)
+ .build();
+
+ private static final ImmutableMap<String, Long> convertToMB =
+ ImmutableMap.<String, Long>builder()
+ .put("1", 1l)
+ .put("1024k", 1l)
+ .put("1024K", 1l)
+ .put("1024kb", 1l)
+ .put("1024Kb", 1l)
+ .put("1024kB", 1l)
+ .put("1024KB", 1l)
+ .put("1m", 1l)
+ .put("1M", 1l)
+ .put("1mb", 1l)
+ .put("1Mb", 1l)
+ .put("1mB", 1l)
+ .put("1MB", 1l)
+ .put("1g", 1024l)
+ .put("1G", 1024l)
+ .put("1gb", 1024l)
+ .put("1gB", 1024l)
+ .put("1Gb", 1024l)
+ .put("1GB", 1024l)
+ .build();
+
+ private static final ImmutableMap<String, Long> convertToGB =
+ ImmutableMap.<String, Long>builder()
+ .put("1", 1l)
+ .put("1024m", 1l)
+ .put("1024M", 1l)
+ .put("1024mb", 1l)
+ .put("1024Mb", 1l)
+ .put("1024mB", 1l)
+ .put("1024MB", 1l)
+ .put("1g", 1l)
+ .put("1G", 1l)
+ .put("1gb", 1l)
+ .put("1gB", 1l)
+ .put("1Gb", 1l)
+ .put("1GB", 1l)
+ .put("1t", 1024l)
+ .put("1T", 1024l)
+ .put("1tb", 1024l)
+ .put("1Tb", 1024l)
+ .put("1tB", 1024l)
+ .put("1TB", 1024l)
+ .build();
+
@Test
- void byteStringAsBytes() {}
+ void byteStringAsBytes() {
+ convertToByte.forEach(
+ (k, v) -> Assertions.assertEquals(opFunction.get("byteStringAsBytes").apply(k), v));
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () -> opFunction.get("byteStringAsBytes").apply("1A"));
+ }
@Test
- void byteStringAsKb() {}
+ void byteStringAsKb() {
+ convertToKB.forEach(
+ (k, v) -> Assertions.assertEquals(opFunction.get("byteStringAsKb").apply(k), v));
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () -> opFunction.get("byteStringAsKb").apply("1a"));
+ }
@Test
- void byteStringAsMb() {}
+ void byteStringAsMb() {
+ convertToMB.forEach(
+ (k, v) -> Assertions.assertEquals(opFunction.get("byteStringAsMb").apply(k), v));
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () -> opFunction.get("byteStringAsMb").apply("1c"));
+ }
@Test
void byteStringAsGb() {
- Long res = ByteTimeUtils.byteStringAsGb("1G");
- Assertions.assertEquals(res, 1);
-
+ convertToGB.forEach(
+ (k, v) -> Assertions.assertEquals(opFunction.get("byteStringAsGb").apply(k), v));
Assertions.assertThrows(
- NullPointerException.class,
- () -> {
- ByteTimeUtils.byteStringAsGb("512");
- });
+ IllegalArgumentException.class, () -> opFunction.get("byteStringAsGb").apply("1C"));
}
}
diff --git a/linkis-commons/linkis-common/src/test/scala/org/apache/linkis/common/utils/VariableUtilsTest.scala b/linkis-commons/linkis-common/src/test/scala/org/apache/linkis/common/utils/VariableUtilsTest.scala
index c0d4ad1d6..e7a105497 100644
--- a/linkis-commons/linkis-common/src/test/scala/org/apache/linkis/common/utils/VariableUtilsTest.scala
+++ b/linkis-commons/linkis-common/src/test/scala/org/apache/linkis/common/utils/VariableUtilsTest.scala
@@ -22,6 +22,8 @@ import org.apache.linkis.common.variable.DateTypeUtils.{getCurHour, getToday}
import java.util
+import scala.collection.mutable
+
import org.junit.jupiter.api.Assertions.assertEquals
import org.junit.jupiter.api.Test
@@ -63,4 +65,19 @@ class VariableUtilsTest {
assertEquals(VariableUtils.replace(sql, "sql", varMap), resSql)
}
+ @Test
+ def testGetCustomVar: Unit = {
+ var scalaCode = "" +
+ "-------@set globalpara=60--------\n" +
+ "--@set globalpara2=66\n" +
+ "select ${globalpara} as globalpara,\n" +
+ "-- ${globalpara1} as globalpara1, \n" +
+ "${globalpara2} as globalpara2;\n"
+ var pythonCode = ""
+
+ val nameAndValue: mutable.Map[String, String] =
+ VariableUtils.getCustomVar(scalaCode, CodeAndRunTypeUtils.LANGUAGE_TYPE_SQL);
+ assertEquals(nameAndValue.size, 2)
+ }
+
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@linkis.apache.org
For additional commands, e-mail: commits-help@linkis.apache.org
[linkis] 03/11: add debug switch params to task
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
peacewong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/linkis.git
commit d245994970df12e07b8d5c8592aab6628e4a28df
Author: peacewong <wp...@gmail.com>
AuthorDate: Fri Sep 22 14:55:15 2023 +0800
add debug switch params to task
---
.../org/apache/linkis/protocol/constants/TaskConstant.java | 1 +
.../scala/org/apache/linkis/protocol/utils/TaskUtils.scala | 11 +++++++++++
2 files changed, 12 insertions(+)
diff --git a/linkis-commons/linkis-protocol/src/main/java/org/apache/linkis/protocol/constants/TaskConstant.java b/linkis-commons/linkis-protocol/src/main/java/org/apache/linkis/protocol/constants/TaskConstant.java
index 42661f828..6eb97c84d 100644
--- a/linkis-commons/linkis-protocol/src/main/java/org/apache/linkis/protocol/constants/TaskConstant.java
+++ b/linkis-commons/linkis-protocol/src/main/java/org/apache/linkis/protocol/constants/TaskConstant.java
@@ -69,6 +69,7 @@ public interface TaskConstant {
String TICKET_ID = "ticketId";
String ENGINE_CONN_TASK_ID = "engineConnTaskId";
String ENGINE_CONN_SUBMIT_TIME = "engineConnSubmitTime";
+ String DEBUG_ENBALE = "debug.enable";
String PARAMS_DATA_SOURCE = "dataSources";
diff --git a/linkis-commons/linkis-protocol/src/main/scala/org/apache/linkis/protocol/utils/TaskUtils.scala b/linkis-commons/linkis-protocol/src/main/scala/org/apache/linkis/protocol/utils/TaskUtils.scala
index 3b94bbdc1..9b2be16ef 100644
--- a/linkis-commons/linkis-protocol/src/main/scala/org/apache/linkis/protocol/utils/TaskUtils.scala
+++ b/linkis-commons/linkis-protocol/src/main/scala/org/apache/linkis/protocol/utils/TaskUtils.scala
@@ -97,4 +97,15 @@ object TaskUtils {
def addLabelsMap(params: util.Map[String, AnyRef], labels: util.Map[String, AnyRef]): Unit =
addMap(params, labels, TaskConstant.LABELS)
+ def isWithDebugInfo(params: util.Map[String, AnyRef]): Boolean = {
+ val debug = getConfigurationMap(params, TaskConstant.PARAMS_CONFIGURATION_STARTUP).get(
+ TaskConstant.DEBUG_ENBALE
+ )
+ if (debug != null && "true".equals(debug.toString)) {
+ true
+ } else {
+ false
+ }
+ }
+
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@linkis.apache.org
For additional commands, e-mail: commits-help@linkis.apache.org
[linkis] 07/11: add version info to linkis-cli
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
peacewong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/linkis.git
commit efc85b3801ff6b34d317873b420a7ff6ca9d8138
Author: peacewong <wp...@gmail.com>
AuthorDate: Wed Oct 4 23:37:18 2023 +0800
add version info to linkis-cli
---
.../linkis-client/linkis-cli/pom.xml | 6 +++
.../apache/linkis/cli/application/CtxBuilder.java | 21 +++++++-
.../cli/application/LinkisClientApplication.java | 41 ++++++++++----
.../linkis/cli/application/constants/CliKeys.java | 2 +
.../cli/application/constants/LinkisKeys.java | 1 +
.../command/template/UniversalCmdTemplate.java | 12 +++++
.../application/interactor/job/help/HelpJob.java | 62 ++++++++++++++++++++++
.../interactor/job/version/VersionJob.java | 48 +++++++++++++++++
.../job/version/VersionJobResult.java} | 33 +++++++++++-
.../application/operator/ujes/LinkisJobOper.java | 4 +-
.../src/main/resources/version.properties | 16 ++++++
11 files changed, 232 insertions(+), 14 deletions(-)
diff --git a/linkis-computation-governance/linkis-client/linkis-cli/pom.xml b/linkis-computation-governance/linkis-client/linkis-cli/pom.xml
index 9d0cbbb84..76723c4ad 100644
--- a/linkis-computation-governance/linkis-client/linkis-cli/pom.xml
+++ b/linkis-computation-governance/linkis-client/linkis-cli/pom.xml
@@ -62,6 +62,12 @@
<build>
<finalName>${project.artifactId}-${project.version}</finalName>
+ <resources>
+ <resource>
+ <filtering>true</filtering>
+ <directory>${basedir}/src/main/resources</directory>
+ </resource>
+ </resources>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
diff --git a/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/CtxBuilder.java b/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/CtxBuilder.java
index 592725f85..07df8b79a 100644
--- a/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/CtxBuilder.java
+++ b/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/CtxBuilder.java
@@ -46,9 +46,11 @@ import org.apache.linkis.cli.application.utils.LoggerManager;
import org.apache.commons.lang3.StringUtils;
+import java.io.*;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Properties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -156,6 +158,23 @@ public class CtxBuilder {
.init();
logger.info("==========std_var============\n" + CliUtils.GSON.toJson(varAccess));
- return new CliCtxImpl(params.getCmdType(), template, varAccess, new HashMap<>());
+ Properties props = new Properties();
+ try (InputStream inputStream =
+ CtxBuilder.class.getClassLoader().getResourceAsStream("version.properties")) {
+ try (InputStreamReader reader = new InputStreamReader(inputStream)) {
+ try (BufferedReader bufferedReader = new BufferedReader(reader)) {
+ props.load(bufferedReader);
+ }
+ }
+ } catch (Exception e) {
+ logger.warn("Failed to load version info", e);
+ }
+
+ String verion = props.getProperty(CliKeys.VERSION);
+
+ Map<String, Object> extraMap = new HashMap<>();
+ extraMap.put(CliKeys.VERSION, verion);
+
+ return new CliCtxImpl(params.getCmdType(), template, varAccess, extraMap);
}
}
diff --git a/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/LinkisClientApplication.java b/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/LinkisClientApplication.java
index f13c7399f..1fb21043a 100644
--- a/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/LinkisClientApplication.java
+++ b/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/LinkisClientApplication.java
@@ -26,9 +26,11 @@ import org.apache.linkis.cli.application.entity.job.JobResult;
import org.apache.linkis.cli.application.exception.CommandException;
import org.apache.linkis.cli.application.interactor.command.CmdTemplateFactory;
import org.apache.linkis.cli.application.interactor.command.template.UniversalCmdTemplate;
+import org.apache.linkis.cli.application.interactor.job.help.HelpJob;
import org.apache.linkis.cli.application.interactor.job.interactive.InteractiveJob;
import org.apache.linkis.cli.application.interactor.job.jobcmd.JobCmdJob;
import org.apache.linkis.cli.application.interactor.job.once.LinkisOnceJob;
+import org.apache.linkis.cli.application.interactor.job.version.VersionJob;
import org.apache.linkis.cli.application.operator.OperManager;
import org.apache.linkis.cli.application.operator.once.OnceOperBuilder;
import org.apache.linkis.cli.application.operator.ujes.LinkisOperBuilder;
@@ -49,6 +51,8 @@ import org.slf4j.LoggerFactory;
public class LinkisClientApplication {
private static Logger logger = LoggerFactory.getLogger(LinkisClientApplication.class);
+ private static boolean showHelp = false;
+
public static void main(String[] args) {
/*
generate template
@@ -64,7 +68,9 @@ public class LinkisClientApplication {
} catch (CommandException e) {
CmdTemplate template = CmdTemplateFactory.getTemplateOri(e.getCmdType());
if (template != null) {
- printHelp(template);
+ HelpInfoModel model = new HelpInfoModel();
+ model.buildModel(ctx.getTemplate());
+ new HelpPresenter().present(model);
}
LoggerManager.getInformationLogger().error("Failed to build CliCtx", e);
System.exit(-1);
@@ -80,7 +86,11 @@ public class LinkisClientApplication {
run job
*/
Job job;
- if (isJobCmd(ctx)) {
+ if (isVersionCmd(ctx)) {
+ job = new VersionJob();
+ } else if (isHelp(ctx)) {
+ job = new HelpJob();
+ } else if (isJobCmd(ctx)) {
job = new JobCmdJob();
} else if (isOnceCmd(ctx)) {
job = new LinkisOnceJob();
@@ -134,21 +144,34 @@ public class LinkisClientApplication {
}
}
- private static void printHelp(CmdTemplate template) {
- HelpInfoModel model = new HelpInfoModel();
- model.buildModel(template);
- new HelpPresenter().present(model);
- }
-
private static void printIndicator(JobResult jobResult) {
if (jobResult.isSuccess()) {
LoggerManager.getPlaintTextLogger().info(CliConstants.SUCCESS_INDICATOR);
} else {
+ LoggerManager.getPlaintTextLogger().info(jobResult.getMessage());
+ StringBuilder b = new StringBuilder();
+ for (Map.Entry<String, String> e : jobResult.getExtraMessage().entrySet()) {
+ b.append(e.getKey()).append(":").append(e.getValue()).append(System.lineSeparator());
+ }
+ LoggerManager.getPlaintTextLogger().info(b.toString());
LoggerManager.getPlaintTextLogger().info(CliConstants.FAILURE_INDICATOR);
- LoggerManager.getPlaintTextLogger().error(jobResult.getMessage());
}
}
+ private static Boolean isHelp(CliCtx ctx) {
+ if (ctx.getVarAccess().hasVar(CliKeys.LINKIS_CLIENT_HELP_OPT)) {
+ return true;
+ }
+ return false;
+ }
+
+ private static Boolean isVersionCmd(CliCtx ctx) {
+ if (ctx.getVarAccess().hasVar(CliKeys.VERSION)) {
+ return true;
+ }
+ return false;
+ }
+
private static Boolean isJobCmd(CliCtx ctx) {
if (ctx.getVarAccess().hasVar(CliKeys.LINKIS_CLIENT_KILL_OPT)
|| ctx.getVarAccess().hasVar(CliKeys.LINKIS_CLIENT_STATUS_OPT)
diff --git a/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/constants/CliKeys.java b/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/constants/CliKeys.java
index 5c04cbcd6..966836e0b 100644
--- a/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/constants/CliKeys.java
+++ b/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/constants/CliKeys.java
@@ -38,6 +38,8 @@ public class CliKeys {
public static final String DEFAULT_CONFIG_FILE_NAME_KEY = "conf.file";
public static final String LINUX_USER_KEY = "user.name";
+ public static final String VERSION = "cli.version";
+
/** Configurable */
/*
execution type
diff --git a/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/constants/LinkisKeys.java b/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/constants/LinkisKeys.java
index 848300313..170f1a8f8 100644
--- a/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/constants/LinkisKeys.java
+++ b/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/constants/LinkisKeys.java
@@ -47,4 +47,5 @@ public class LinkisKeys {
public static final String KEY_YARN_QUEUE = "wds.linkis.rm.yarnqueue";
public static final String KEY_HIVE_RESULT_DISPLAY_TBALE =
"hive.resultset.use.unique.column.names";
+ public static final String CLI_VERSION = "cli.version";
}
diff --git a/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/interactor/command/template/UniversalCmdTemplate.java b/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/interactor/command/template/UniversalCmdTemplate.java
index f761a6a47..83b42032c 100644
--- a/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/interactor/command/template/UniversalCmdTemplate.java
+++ b/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/interactor/command/template/UniversalCmdTemplate.java
@@ -271,6 +271,15 @@ public class UniversalCmdTemplate extends AbstractCmdTemplate implements Cloneab
"specify jobContent map. Input format: -jobContentMap key1=value1 -jobContentMap key2=value2",
true);
+ protected Flag versionFlag =
+ flag(
+ CliKeys.VERSION,
+ CliKeys.VERSION,
+ new String[] {"--version"},
+ "show version",
+ true,
+ false);
+
protected Parameter<String[]> argumentsParas =
parameter(
CliKeys.LINKIS_CLIENT_COMMON,
@@ -286,6 +295,9 @@ public class UniversalCmdTemplate extends AbstractCmdTemplate implements Cloneab
@Override
public void checkParams() throws CommandException {
+ if (versionFlag.hasVal()) {
+ return;
+ }
int cnt = 0;
if (statusOpt.hasVal()) {
cnt++;
diff --git a/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/interactor/job/help/HelpJob.java b/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/interactor/job/help/HelpJob.java
new file mode 100644
index 000000000..943c54ed6
--- /dev/null
+++ b/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/interactor/job/help/HelpJob.java
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.linkis.cli.application.interactor.job.help;
+
+import org.apache.linkis.cli.application.entity.context.CliCtx;
+import org.apache.linkis.cli.application.entity.job.Job;
+import org.apache.linkis.cli.application.entity.job.JobResult;
+import org.apache.linkis.cli.application.present.HelpPresenter;
+import org.apache.linkis.cli.application.present.model.HelpInfoModel;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class HelpJob implements Job {
+ private CliCtx ctx;
+
+ @Override
+ public void build(CliCtx ctx) {
+ this.ctx = ctx;
+ }
+
+ @Override
+ public JobResult run() {
+ HelpInfoModel model = new HelpInfoModel();
+ model.buildModel(ctx.getTemplate());
+ new HelpPresenter().present(model);
+ return new JobResult() {
+ @Override
+ public Boolean isSuccess() {
+ return true;
+ }
+
+ @Override
+ public String getMessage() {
+ return "";
+ }
+
+ @Override
+ public Map<String, String> getExtraMessage() {
+ return new HashMap<>();
+ }
+ };
+ }
+
+ @Override
+ public void onDestroy() {}
+}
diff --git a/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/interactor/job/version/VersionJob.java b/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/interactor/job/version/VersionJob.java
new file mode 100644
index 000000000..599f97904
--- /dev/null
+++ b/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/interactor/job/version/VersionJob.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.linkis.cli.application.interactor.job.version;
+
+import org.apache.linkis.cli.application.constants.CliKeys;
+import org.apache.linkis.cli.application.entity.context.CliCtx;
+import org.apache.linkis.cli.application.entity.job.Job;
+import org.apache.linkis.cli.application.entity.job.JobResult;
+import org.apache.linkis.cli.application.utils.LoggerManager;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class VersionJob implements Job {
+ private CliCtx ctx;
+
+ @Override
+ public void build(CliCtx cliCtx) {
+ this.ctx = cliCtx;
+ }
+
+ @Override
+ public JobResult run() {
+ String version = (String) ctx.getExtraMap().get(CliKeys.VERSION);
+ Map<String, String> extraMap = new HashMap<>();
+ extraMap.put(CliKeys.VERSION, version);
+ LoggerManager.getPlaintTextLogger().info("Version=" + version);
+ return new VersionJobResult(true, "ok", extraMap);
+ }
+
+ @Override
+ public void onDestroy() {}
+}
diff --git a/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/entity/version/Version.java b/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/interactor/job/version/VersionJobResult.java
similarity index 53%
rename from linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/entity/version/Version.java
rename to linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/interactor/job/version/VersionJobResult.java
index b0bcd8b6d..e2f12cd7c 100644
--- a/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/entity/version/Version.java
+++ b/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/interactor/job/version/VersionJobResult.java
@@ -15,6 +15,35 @@
* limitations under the License.
*/
-package org.apache.linkis.cli.application.entity.version;
+package org.apache.linkis.cli.application.interactor.job.version;
-public interface Version {}
+import org.apache.linkis.cli.application.entity.job.JobResult;
+
+import java.util.Map;
+
+public class VersionJobResult implements JobResult {
+ private Boolean success;
+ private String message;
+ private Map<String, String> extraMsg;
+
+ public VersionJobResult(Boolean success, String message, Map<String, String> extraMsg) {
+ this.success = success;
+ this.message = message;
+ this.extraMsg = extraMsg;
+ }
+
+ @Override
+ public Boolean isSuccess() {
+ return success;
+ }
+
+ @Override
+ public String getMessage() {
+ return message;
+ }
+
+ @Override
+ public Map<String, String> getExtraMessage() {
+ return extraMsg;
+ }
+}
diff --git a/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/operator/ujes/LinkisJobOper.java b/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/operator/ujes/LinkisJobOper.java
index f696c8674..bfebf62c7 100644
--- a/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/operator/ujes/LinkisJobOper.java
+++ b/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/operator/ujes/LinkisJobOper.java
@@ -269,7 +269,7 @@ public class LinkisJobOper implements JobOper {
}
String msg =
MessageFormat.format(
- "Get job info failed. retry time : {0}/{1}. taskID={0}, Reason: {1}",
+ "Get job info failed. retry time : {0}/{1}. taskID={2}, Reason: {3}",
retryTime, MAX_RETRY_TIME, taskID, reason);
logger.debug(
@@ -299,7 +299,7 @@ public class LinkisJobOper implements JobOper {
if (jobInfoResult == null) {
reason = "JobInfoResult is null";
} else {
- reason = "server returns non-zero status-code";
+ reason = "server returns non-zero status-code. ";
reason += jobInfoResult.getMessage();
}
String msg =
diff --git a/linkis-computation-governance/linkis-client/linkis-cli/src/main/resources/version.properties b/linkis-computation-governance/linkis-client/linkis-cli/src/main/resources/version.properties
new file mode 100644
index 000000000..0da37e5c0
--- /dev/null
+++ b/linkis-computation-governance/linkis-client/linkis-cli/src/main/resources/version.properties
@@ -0,0 +1,16 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+# http://www.apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+cli.version=${project.version}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@linkis.apache.org
For additional commands, e-mail: commits-help@linkis.apache.org
[linkis] 09/11: support null
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
peacewong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/linkis.git
commit 88d9e1be7391c790b7bb0fee70b25494df88caae
Author: peacewong <wp...@gmail.com>
AuthorDate: Wed Oct 4 23:38:44 2023 +0800
support null
---
.../org/apache/linkis/ujes/client/request/ResultSetAction.scala | 9 +++++++++
1 file changed, 9 insertions(+)
diff --git a/linkis-computation-governance/linkis-client/linkis-computation-client/src/main/scala/org/apache/linkis/ujes/client/request/ResultSetAction.scala b/linkis-computation-governance/linkis-client/linkis-computation-client/src/main/scala/org/apache/linkis/ujes/client/request/ResultSetAction.scala
index 6b41b4c62..9eb748691 100644
--- a/linkis-computation-governance/linkis-client/linkis-computation-client/src/main/scala/org/apache/linkis/ujes/client/request/ResultSetAction.scala
+++ b/linkis-computation-governance/linkis-client/linkis-computation-client/src/main/scala/org/apache/linkis/ujes/client/request/ResultSetAction.scala
@@ -35,6 +35,9 @@ object ResultSetAction {
private var pageSize: Int = _
private var charset: String = Configuration.BDP_ENCODING.getValue
+ // default value is :org.apache.linkis.storage.domain.Dolphin.LINKIS_NULL
+ private var nullValue: String = "LINKIS_NULL"
+
def setUser(user: String): Builder = {
this.user = user
this
@@ -60,6 +63,11 @@ object ResultSetAction {
this
}
+ def setNullValue(nullValue: String): Builder = {
+ this.nullValue = nullValue
+ this
+ }
+
def build(): ResultSetAction = {
if (user == null) throw new UJESClientBuilderException("user is needed!")
if (path == null) throw new UJESClientBuilderException("path is needed!")
@@ -68,6 +76,7 @@ object ResultSetAction {
if (page > 0) resultSetAction.setParameter("page", page)
if (pageSize > 0) resultSetAction.setParameter("pageSize", pageSize)
resultSetAction.setParameter("charset", charset)
+ resultSetAction.setParameter("nullValue", nullValue)
resultSetAction.setUser(user)
resultSetAction
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@linkis.apache.org
For additional commands, e-mail: commits-help@linkis.apache.org
[linkis] 08/11: optimize jdbc-driver module
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
peacewong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/linkis.git
commit c625de68e300eaedc535987ab1158716f664d11a
Author: peacewong <wp...@gmail.com>
AuthorDate: Wed Oct 4 23:38:30 2023 +0800
optimize jdbc-driver module
---
.../org/apache/linkis/ujes/jdbc/UJESSQLDriver.java | 4 +---
.../apache/linkis/ujes/jdbc/utils/JDBCUtils.java | 10 +++++++++
.../linkis/ujes/jdbc/LinkisSQLConnection.scala | 19 +++++++++++++++++
.../linkis/ujes/jdbc/UJESClientFactory.scala | 1 -
.../linkis/ujes/jdbc/UJESSQLDriverMain.scala | 24 +++++++++++-----------
.../apache/linkis/ujes/jdbc/UJESSQLResultSet.scala | 14 ++++++-------
.../jdbc/hook/JDBCDriverPreExecutionHook.scala | 3 +--
7 files changed, 49 insertions(+), 26 deletions(-)
diff --git a/linkis-computation-governance/linkis-jdbc-driver/src/main/java/org/apache/linkis/ujes/jdbc/UJESSQLDriver.java b/linkis-computation-governance/linkis-jdbc-driver/src/main/java/org/apache/linkis/ujes/jdbc/UJESSQLDriver.java
index 392f4dd09..c12e2791b 100644
--- a/linkis-computation-governance/linkis-jdbc-driver/src/main/java/org/apache/linkis/ujes/jdbc/UJESSQLDriver.java
+++ b/linkis-computation-governance/linkis-jdbc-driver/src/main/java/org/apache/linkis/ujes/jdbc/UJESSQLDriver.java
@@ -48,9 +48,7 @@ public class UJESSQLDriver extends UJESSQLDriverMain implements Driver {
static String TOKEN_VALUE = "value";
static String PASSWORD = "password";
static boolean TABLEAU_SERVER = false;
- static String LIMIT_ENABLED = "true";
- static String LIMIT = "limit";
-
+ static String FIXED_SESSION = "fixedSession";
static String VERSION = "version";
static int DEFAULT_VERSION = 1;
static String MAX_CONNECTION_SIZE = "maxConnectionSize";
diff --git a/linkis-computation-governance/linkis-jdbc-driver/src/main/java/org/apache/linkis/ujes/jdbc/utils/JDBCUtils.java b/linkis-computation-governance/linkis-jdbc-driver/src/main/java/org/apache/linkis/ujes/jdbc/utils/JDBCUtils.java
index 8f7dcfed1..3e1e7e318 100644
--- a/linkis-computation-governance/linkis-jdbc-driver/src/main/java/org/apache/linkis/ujes/jdbc/utils/JDBCUtils.java
+++ b/linkis-computation-governance/linkis-jdbc-driver/src/main/java/org/apache/linkis/ujes/jdbc/utils/JDBCUtils.java
@@ -17,10 +17,16 @@
package org.apache.linkis.ujes.jdbc.utils;
+import org.apache.linkis.common.utils.Utils;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
public class JDBCUtils {
private static final char SEARCH_STRING_ESCAPE = '\\';
+ public static final AtomicInteger idCreator = new AtomicInteger();
+
public static String convertPattern(final String pattern) {
if (pattern == null) {
return ".*";
@@ -50,4 +56,8 @@ public class JDBCUtils {
return result.toString();
}
}
+
+ public static String getUniqId() {
+ return Utils.getLocalHostname() + "_" + JDBCUtils.idCreator.getAndIncrement();
+ }
}
diff --git a/linkis-computation-governance/linkis-jdbc-driver/src/main/scala/org/apache/linkis/ujes/jdbc/LinkisSQLConnection.scala b/linkis-computation-governance/linkis-jdbc-driver/src/main/scala/org/apache/linkis/ujes/jdbc/LinkisSQLConnection.scala
index e111615ce..99bb24773 100644
--- a/linkis-computation-governance/linkis-jdbc-driver/src/main/scala/org/apache/linkis/ujes/jdbc/LinkisSQLConnection.scala
+++ b/linkis-computation-governance/linkis-jdbc-driver/src/main/scala/org/apache/linkis/ujes/jdbc/LinkisSQLConnection.scala
@@ -25,6 +25,7 @@ import org.apache.linkis.ujes.client.UJESClient
import org.apache.linkis.ujes.client.request.JobSubmitAction
import org.apache.linkis.ujes.client.response.JobExecuteResult
import org.apache.linkis.ujes.jdbc.UJESSQLDriverMain._
+import org.apache.linkis.ujes.jdbc.utils.JDBCUtils
import org.apache.commons.lang3.StringUtils
@@ -99,6 +100,18 @@ class LinkisSQLConnection(private[jdbc] val ujesClient: UJESClient, props: Prope
private[jdbc] val serverURL = props.getProperty("URL")
+ private[jdbc] val fixedSessionEnabled =
+ if (
+ props
+ .containsKey(FIXED_SESSION) && "true".equalsIgnoreCase(props.getProperty(FIXED_SESSION))
+ ) {
+ true
+ } else {
+ false
+ }
+
+ private val connectionId = JDBCUtils.getUniqId()
+
private val labelMap: util.Map[String, AnyRef] = new util.HashMap[String, AnyRef]
private val startupParams: util.Map[String, AnyRef] = new util.HashMap[String, AnyRef]
@@ -445,6 +458,10 @@ class LinkisSQLConnection(private[jdbc] val ujesClient: UJESClient, props: Prope
labelMap.put(LabelKeyConstant.ENGINE_TYPE_KEY, engineTypeLabel.getStringValue)
labelMap.put(LabelKeyConstant.USER_CREATOR_TYPE_KEY, s"$user-$creator")
labelMap.put(LabelKeyConstant.CODE_TYPE_KEY, engineToCodeType(engineTypeLabel.getEngineType))
+ if (fixedSessionEnabled) {
+ labelMap.put(LabelKeyConstant.FIXED_EC_KEY, connectionId)
+ logger.info("Fixed session is enable session id is {}", connectionId)
+ }
val jobSubmitAction = JobSubmitAction.builder
.addExecuteCode(code)
@@ -463,4 +480,6 @@ class LinkisSQLConnection(private[jdbc] val ujesClient: UJESClient, props: Prope
result
}
+ override def toString: String = "LinkisConnection_" + connectionId
+
}
diff --git a/linkis-computation-governance/linkis-jdbc-driver/src/main/scala/org/apache/linkis/ujes/jdbc/UJESClientFactory.scala b/linkis-computation-governance/linkis-jdbc-driver/src/main/scala/org/apache/linkis/ujes/jdbc/UJESClientFactory.scala
index 517f8b07a..bea24181a 100644
--- a/linkis-computation-governance/linkis-jdbc-driver/src/main/scala/org/apache/linkis/ujes/jdbc/UJESClientFactory.scala
+++ b/linkis-computation-governance/linkis-jdbc-driver/src/main/scala/org/apache/linkis/ujes/jdbc/UJESClientFactory.scala
@@ -66,7 +66,6 @@ object UJESClientFactory extends Logging {
clientConfigBuilder.setAuthenticationStrategy(new StaticAuthenticationStrategy())
clientConfigBuilder.readTimeout(100000)
clientConfigBuilder.maxConnectionSize(20)
- clientConfigBuilder.readTimeout(10000)
val params = props.getProperty(PARAMS)
var versioned = false
if (StringUtils.isNotBlank(params)) {
diff --git a/linkis-computation-governance/linkis-jdbc-driver/src/main/scala/org/apache/linkis/ujes/jdbc/UJESSQLDriverMain.scala b/linkis-computation-governance/linkis-jdbc-driver/src/main/scala/org/apache/linkis/ujes/jdbc/UJESSQLDriverMain.scala
index 6296f399e..ab2f6dda1 100644
--- a/linkis-computation-governance/linkis-jdbc-driver/src/main/scala/org/apache/linkis/ujes/jdbc/UJESSQLDriverMain.scala
+++ b/linkis-computation-governance/linkis-jdbc-driver/src/main/scala/org/apache/linkis/ujes/jdbc/UJESSQLDriverMain.scala
@@ -29,10 +29,10 @@ import java.sql.{
DriverPropertyInfo,
SQLFeatureNotSupportedException
}
-import java.util.Properties
+import java.util.{Locale, Properties}
import java.util.logging.Logger
-import scala.collection.JavaConversions
+import scala.collection.JavaConverters._
class UJESSQLDriverMain extends Driver with Logging {
@@ -72,9 +72,8 @@ class UJESSQLDriverMain extends Driver with Logging {
case Array(TOKEN_VALUE, value) =>
props.setProperty(TOKEN_VALUE, value)
false
- case Array(LIMIT, value) =>
- props.setProperty(LIMIT, value)
- UJESSQLDriverMain.LIMIT_ENABLED = value.toLowerCase()
+ case Array(FIXED_SESSION, value) =>
+ props.setProperty(FIXED_SESSION, value)
false
case Array(key, _) =>
if (StringUtils.isBlank(key)) {
@@ -138,8 +137,7 @@ object UJESSQLDriverMain {
val TOKEN_VALUE = UJESSQLDriver.TOKEN_VALUE
val PASSWORD = UJESSQLDriver.PASSWORD
val TABLEAU_SERVER = UJESSQLDriver.TABLEAU_SERVER
- val LIMIT = UJESSQLDriver.LIMIT
- var LIMIT_ENABLED = UJESSQLDriver.LIMIT_ENABLED
+ val FIXED_SESSION = UJESSQLDriver.FIXED_SESSION
val VERSION = UJESSQLDriver.VERSION
val DEFAULT_VERSION = UJESSQLDriver.DEFAULT_VERSION
@@ -157,8 +155,7 @@ object UJESSQLDriverMain {
connectionParams: String,
variableMap: java.util.Map[String, Any]
): String = {
- val variables = JavaConversions
- .mapAsScalaMap(variableMap)
+ val variables = variableMap.asScala
.map(kv => VARIABLE_HEADER + kv._1 + KV_SPLIT + kv._2)
.mkString(PARAM_SPLIT)
if (StringUtils.isNotBlank(connectionParams)) connectionParams + PARAM_SPLIT + variables
@@ -186,17 +183,20 @@ object UJESSQLDriverMain {
): String = {
val sb = new StringBuilder
if (StringUtils.isNotBlank(version)) sb.append(VERSION).append(KV_SPLIT).append(version)
- if (maxConnectionSize > 0)
+ if (maxConnectionSize > 0) {
sb.append(PARAM_SPLIT).append(MAX_CONNECTION_SIZE).append(KV_SPLIT).append(maxConnectionSize)
- if (readTimeout > 0)
+ }
+ if (readTimeout > 0) {
sb.append(PARAM_SPLIT).append(READ_TIMEOUT).append(KV_SPLIT).append(readTimeout)
+ }
if (enableDiscovery) {
sb.append(PARAM_SPLIT).append(ENABLE_DISCOVERY).append(KV_SPLIT).append(enableDiscovery)
- if (enableLoadBalancer)
+ if (enableLoadBalancer) {
sb.append(PARAM_SPLIT)
.append(ENABLE_LOADBALANCER)
.append(KV_SPLIT)
.append(enableLoadBalancer)
+ }
}
if (sb.startsWith(PARAM_SPLIT)) sb.toString.substring(PARAM_SPLIT.length) else sb.toString
}
diff --git a/linkis-computation-governance/linkis-jdbc-driver/src/main/scala/org/apache/linkis/ujes/jdbc/UJESSQLResultSet.scala b/linkis-computation-governance/linkis-jdbc-driver/src/main/scala/org/apache/linkis/ujes/jdbc/UJESSQLResultSet.scala
index 5df56733e..0ed47925c 100644
--- a/linkis-computation-governance/linkis-jdbc-driver/src/main/scala/org/apache/linkis/ujes/jdbc/UJESSQLResultSet.scala
+++ b/linkis-computation-governance/linkis-jdbc-driver/src/main/scala/org/apache/linkis/ujes/jdbc/UJESSQLResultSet.scala
@@ -193,7 +193,7 @@ class UJESSQLResultSet(
if (metaData == null) init()
currentRowCursor += 1
if (null == resultSetRow || currentRowCursor > resultSetRow.size() - 1) {
- if (UJESSQLDriverMain.LIMIT_ENABLED.equals("false") && !isCompleted) {
+ if (!isCompleted) {
updateResultSet()
if (isCompleted) {
return false
@@ -242,7 +242,7 @@ class UJESSQLResultSet(
} else {
dataType.toLowerCase(Locale.getDefault) match {
case null => throw new LinkisSQLException(LinkisSQLErrorCode.METADATA_EMPTY)
- case "string" => value.toString
+ case "char" | "varchar" | "nvarchar" | "string" => value
case "short" => value.toShort
case "int" => value.toInt
case "long" => value.toLong
@@ -250,11 +250,8 @@ class UJESSQLResultSet(
case "double" => value.toDouble
case "boolean" => value.toBoolean
case "byte" => value.toByte
- case "char" => value.toString
- case "timestamp" => value.toString
- case "varchar" => value.toString
- case "nvarchar" => value.toString
- case "date" => value.toString
+ case "timestamp" => value
+ case "date" => value
case "bigint" => value.toLong
case "decimal" => value.toDouble
case "array" => value.toArray
@@ -974,8 +971,9 @@ class UJESSQLResultSet(
logger.info(s"the value of value is $value and the value of localTimeZone is $localTimeZone")
if (wasNull()) {
null
- } else
+ } else {
new Timestamp(TIMESTAMP_FORMATTER.withZone(localTimeZone).parseMillis(String.valueOf(value)))
+ }
}
override def getTimestamp(columnIndex: Int, cal: Calendar): Timestamp = {
diff --git a/linkis-computation-governance/linkis-jdbc-driver/src/main/scala/org/apache/linkis/ujes/jdbc/hook/JDBCDriverPreExecutionHook.scala b/linkis-computation-governance/linkis-jdbc-driver/src/main/scala/org/apache/linkis/ujes/jdbc/hook/JDBCDriverPreExecutionHook.scala
index ed8936cc7..c7de7d373 100644
--- a/linkis-computation-governance/linkis-jdbc-driver/src/main/scala/org/apache/linkis/ujes/jdbc/hook/JDBCDriverPreExecutionHook.scala
+++ b/linkis-computation-governance/linkis-jdbc-driver/src/main/scala/org/apache/linkis/ujes/jdbc/hook/JDBCDriverPreExecutionHook.scala
@@ -34,8 +34,7 @@ object JDBCDriverPreExecutionHook extends Logging {
val hooks = new ArrayBuffer[JDBCDriverPreExecutionHook]()
CommonVars(
"wds.linkis.jdbc.pre.hook.class",
- "org.apache.linkis.ujes.jdbc.hook.impl.TableauPreExecutionHook," +
- "org.apache.linkis.ujes.jdbc.hook.impl.NoLimitExecutionHook"
+ "org.apache.linkis.ujes.jdbc.hook.impl.TableauPreExecutionHook"
).getValue.split(",") foreach { hookStr =>
Utils.tryCatch {
val clazz = Class.forName(hookStr.trim)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@linkis.apache.org
For additional commands, e-mail: commits-help@linkis.apache.org