From 94d9a7c2984aae2493d3fca3c05b0c158f54a3fe Mon Sep 17 00:00:00 2001 From: Hongze Zhang Date: Tue, 25 May 2021 14:15:24 +0800 Subject: [PATCH 1/2] [NSE-337] UDF: Add test case for validating basic row-based udf --- .../scala/com/intel/oap/tpc/ds/TPCDSSuite.scala | 13 +++++++++++-- .../test/scala/com/intel/oap/tpc/h/TPCHSuite.scala | 2 +- 2 files changed, 12 insertions(+), 3 deletions(-) diff --git a/native-sql-engine/core/src/test/scala/com/intel/oap/tpc/ds/TPCDSSuite.scala b/native-sql-engine/core/src/test/scala/com/intel/oap/tpc/ds/TPCDSSuite.scala index 128f68bb7..58577c684 100644 --- a/native-sql-engine/core/src/test/scala/com/intel/oap/tpc/ds/TPCDSSuite.scala +++ b/native-sql-engine/core/src/test/scala/com/intel/oap/tpc/ds/TPCDSSuite.scala @@ -26,7 +26,7 @@ import org.apache.spark.sql.test.SharedSparkSession class TPCDSSuite extends QueryTest with SharedSparkSession { private val MAX_DIRECT_MEMORY = "6g" - private val TPCDS_QUERIES_RESOURCE = "tpcds-queries" + private val TPCDS_QUERIES_RESOURCE = "tpcds" private val TPCDS_WRITE_PATH = "/tmp/tpcds-generated" private var runner: TPCRunner = _ @@ -60,7 +60,7 @@ class TPCDSSuite extends QueryTest with SharedSparkSession { super.beforeAll() LogManager.getRootLogger.setLevel(Level.WARN) val tGen = new TPCDSTableGen(spark, 0.01D, TPCDS_WRITE_PATH) - tGen.gen() +// tGen.gen() tGen.createTables() runner = new TPCRunner(spark, TPCDS_QUERIES_RESOURCE) } @@ -136,6 +136,15 @@ class TPCDSSuite extends QueryTest with SharedSparkSession { df.explain() df.show() } + + test("simple UDF") { + spark.udf.register("strLenScala", + (s: String) => Option(s).map(_.length).orElse(Option(0)).get) + val df = spark.sql("SELECT i_item_sk, i_item_desc, strLenScala(i_item_desc) FROM " + + "item LIMIT 100") + df.explain() + df.show() + } } object TPCDSSuite { diff --git a/native-sql-engine/core/src/test/scala/com/intel/oap/tpc/h/TPCHSuite.scala b/native-sql-engine/core/src/test/scala/com/intel/oap/tpc/h/TPCHSuite.scala index a8fc06e86..82ac63e46 100644 --- a/native-sql-engine/core/src/test/scala/com/intel/oap/tpc/h/TPCHSuite.scala +++ b/native-sql-engine/core/src/test/scala/com/intel/oap/tpc/h/TPCHSuite.scala @@ -41,7 +41,7 @@ import scala.collection.mutable.ArrayBuffer class TPCHSuite extends QueryTest with SharedSparkSession { private val MAX_DIRECT_MEMORY = "6g" - private val TPCH_QUERIES_RESOURCE = "tpch-queries" + private val TPCH_QUERIES_RESOURCE = "tpch" private val TPCH_WRITE_PATH = "/tmp/tpch-generated" private var runner: TPCRunner = _ From 4c74ba3b27a048469c5516ce3436ac066a40be7c Mon Sep 17 00:00:00 2001 From: Hongze Zhang Date: Tue, 25 May 2021 14:16:24 +0800 Subject: [PATCH 2/2] fixup --- .../core/src/test/scala/com/intel/oap/tpc/ds/TPCDSSuite.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/native-sql-engine/core/src/test/scala/com/intel/oap/tpc/ds/TPCDSSuite.scala b/native-sql-engine/core/src/test/scala/com/intel/oap/tpc/ds/TPCDSSuite.scala index 58577c684..cf753fc93 100644 --- a/native-sql-engine/core/src/test/scala/com/intel/oap/tpc/ds/TPCDSSuite.scala +++ b/native-sql-engine/core/src/test/scala/com/intel/oap/tpc/ds/TPCDSSuite.scala @@ -60,7 +60,7 @@ class TPCDSSuite extends QueryTest with SharedSparkSession { super.beforeAll() LogManager.getRootLogger.setLevel(Level.WARN) val tGen = new TPCDSTableGen(spark, 0.01D, TPCDS_WRITE_PATH) -// tGen.gen() + tGen.gen() tGen.createTables() runner = new TPCRunner(spark, TPCDS_QUERIES_RESOURCE) }