use testcontainers

apache · Jan 5, 2022 · 3e4f778 · 3e4f778
1 parent 3cd4d94
commit 3e4f778
Show file tree

Hide file tree

Showing 6 changed files with 48 additions and 41 deletions.
diff --git a/externals/kyuubi-trino-engine/pom.xml b/externals/kyuubi-trino-engine/pom.xml
@@ -65,15 +65,21 @@
         </dependency>
 
         <dependency>
-            <groupId>javax.activation</groupId>
-            <artifactId>activation</artifactId>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>testcontainers</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <!-- https://mvnrepository.com/artifact/org.testcontainers/trino -->
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>trino</artifactId>
             <scope>test</scope>
         </dependency>
 
         <dependency>
             <groupId>io.trino</groupId>
-            <artifactId>trino-tests</artifactId>
-            <type>test-jar</type>
+            <artifactId>trino-jdbc</artifactId>
             <scope>test</scope>
         </dependency>
 

diff --git a/...ls/kyuubi-trino-engine/src/main/scala/org/apache/kyuubi/engine/trino/TrinoStatement.scala b/...ls/kyuubi-trino-engine/src/main/scala/org/apache/kyuubi/engine/trino/TrinoStatement.scala
@@ -74,7 +74,7 @@ class TrinoStatement(trinoContext: TrinoContext, sql: String) {
   def execute(): Iterable[List[Any]] = {
     val rowQueue = new ArrayBlockingQueue[List[Any]](MAX_QUEUED_ROWS)
 
-    val future = Future[Unit] {
+    val dataProcessing = Future[Unit] {
       while (trino.isRunning) {
         val data = trino.currentData().getData()
         if (data != null) {
@@ -84,7 +84,7 @@ class TrinoStatement(trinoContext: TrinoContext, sql: String) {
         trino.advance()
       }
     }
-    future.onComplete {
+    dataProcessing.onComplete {
       case _ => putOrThrow(rowQueue, END_TOKEN)
     }
 
@@ -93,7 +93,7 @@ class TrinoStatement(trinoContext: TrinoContext, sql: String) {
     val result = ArrayBuffer[List[Any]]()
     try {
       breakable {
-        while (!future.isCompleted) {
+        while (!dataProcessing.isCompleted) {
           val atEnd = drainDetectingEnd(rowQueue, rowBuffer, MAX_BUFFERED_ROWS, END_TOKEN)
           if (!atEnd) {
             // Flush if needed

diff --git a/...kyuubi-trino-engine/src/test/scala/org/apache/kyuubi/engine/trino/TrinoContextSuite.scala b/...kyuubi-trino-engine/src/test/scala/org/apache/kyuubi/engine/trino/TrinoContextSuite.scala
@@ -17,7 +17,7 @@
 
 package org.apache.kyuubi.engine.trino
 
-class TrinoContextSuite extends WithTrinoLocalServer {
+class TrinoContextSuite extends WithTrinoContainerServer {
 
   test("set current schema") {
     val trinoContext = TrinoContext(httpClient, session)

diff --git a/...uubi-trino-engine/src/test/scala/org/apache/kyuubi/engine/trino/TrinoStatementSuite.scala b/...uubi-trino-engine/src/test/scala/org/apache/kyuubi/engine/trino/TrinoStatementSuite.scala
@@ -19,10 +19,9 @@ package org.apache.kyuubi.engine.trino
 
 import org.apache.kyuubi.KyuubiSQLException
 
-class TrinoStatementSuite extends WithTrinoLocalServer {
+class TrinoStatementSuite extends WithTrinoContainerServer {
 
   test("test query") {
-    info(connectUrl)
     val trinoStatement = TrinoStatement(TrinoContext(httpClient, session), "select 1")
     val schema = trinoStatement.getColumns
     val resultSet = trinoStatement.execute()
@@ -43,11 +42,11 @@ class TrinoStatementSuite extends WithTrinoLocalServer {
 
   test("test update session") {
     val trinoStatement = TrinoStatement(TrinoContext(httpClient, session), "select 1")
-    val schema = trinoStatement.getColumns
+    val schema2 = trinoStatement.getColumns
 
-    assert(schema.size === 1)
-    assert(schema(0).getName === "_col0")
-    assert("tiny" === trinoStatement.getCurrentDatabase)
+    assert(schema2.size === 1)
+    assert(schema2(0).getName === "_col0")
+    assert(this.schema === trinoStatement.getCurrentDatabase)
 
     val trinoStatement2 = TrinoStatement(TrinoContext(httpClient, session), "use sf1")
     trinoStatement2.execute()

diff --git a/...i/engine/trino/WithTrinoLocalServer.scala → ...gine/trino/WithTrinoContainerServer.scala b/...i/engine/trino/WithTrinoLocalServer.scala → ...gine/trino/WithTrinoContainerServer.scala
@@ -28,43 +28,43 @@ import scala.collection.JavaConverters._
 import io.airlift.units.Duration
 import io.trino.client.ClientSelectedRole
 import io.trino.client.ClientSession
-import io.trino.testing.DistributedQueryRunner
-import io.trino.tests.tpch.TpchQueryRunnerBuilder
 import okhttp3.OkHttpClient
+import org.testcontainers.containers.TrinoContainer
 
 import org.apache.kyuubi.KyuubiFunSuite
 
-trait WithTrinoLocalServer extends KyuubiFunSuite {
+trait WithTrinoContainerServer extends KyuubiFunSuite {
 
-  private var queryRunner: DistributedQueryRunner = _
+  final val IMAGE_VERSION = 363
+  final val DOCKER_IMAGE_NAME = s"trinodb/trino:${IMAGE_VERSION}"
+
+  val trino = new TrinoContainer(DOCKER_IMAGE_NAME)
+
+  protected val catalog = "tpch"
+  protected val schema = "tiny"
 
   override def beforeAll(): Unit = {
-    queryRunner = TpchQueryRunnerBuilder.builder()
-      .setExtraProperties(
-        Map(
-          "http-server.http.port" -> "8080",
-          "sql.default-catalog" -> "tpch",
-          "sql.default-schema" -> "tiny").asJava).build()
+    trino.start()
     super.beforeAll()
   }
 
   override def afterAll(): Unit = {
-    queryRunner.close()
+    trino.stop()
     super.afterAll()
   }
 
-  lazy val connectUrl = queryRunner.getCoordinator.getBaseUrl
+  lazy val connectionUrl = trino.getJdbcUrl.replace("jdbc:trino", "http")
 
   lazy val session = new ClientSession(
-    URI.create(connectUrl.toString),
+    URI.create(connectionUrl),
     "kyuubi_test",
     Optional.empty(),
     "kyuubi",
     Optional.empty(),
     Set[String]().asJava,
     null,
-    "tpch",
-    "tiny",
+    catalog,
+    schema,
     null,
     ZoneId.systemDefault(),
     Locale.getDefault,

diff --git a/pom.xml b/pom.xml
@@ -137,6 +137,7 @@
         <swagger.version>2.1.11</swagger.version>
         <swagger.scala.module.version>2.5.2</swagger.scala.module.version>
         <swagger-ui.version>4.1.3</swagger-ui.version>
+        <testcontainers.version>1.16.2</testcontainers.version>
         <trino.client.version>363</trino.client.version>
         <zookeeper.version>3.4.14</zookeeper.version>
 
@@ -454,19 +455,20 @@
 
             <dependency>
                 <groupId>io.trino</groupId>
-                <artifactId>trino-tests</artifactId>
+                <artifactId>trino-jdbc</artifactId>
                 <version>${trino.client.version}</version>
-                <type>test-jar</type>
-                <exclusions>
-                    <exclusion>
-                        <groupId>org.slf4j</groupId>
-                        <artifactId>slf4j-jdk14</artifactId>
-                    </exclusion>
-                    <exclusion>
-                        <groupId>org.slf4j</groupId>
-                        <artifactId>log4j-over-slf4j</artifactId>
-                    </exclusion>
-                </exclusions>
+            </dependency>
+
+            <dependency>
+                <groupId>org.testcontainers</groupId>
+                <artifactId>testcontainers</artifactId>
+                <version>${testcontainers.version}</version>
+            </dependency>
+
+            <dependency>
+                <groupId>org.testcontainers</groupId>
+                <artifactId>trino</artifactId>
+                <version>${testcontainers.version}</version>
             </dependency>
 
             <dependency>