|
17 | 17 |
|
18 | 18 | package org.apache.spark.sql.hive.execution
|
19 | 19 |
|
20 |
| -import org.apache.spark.sql.hive.test.TestHive |
21 |
| -import org.apache.hadoop.conf.Configuration |
22 |
| -import org.apache.spark.SparkContext._ |
| 20 | +import java.io.{DataOutput, DataInput} |
23 | 21 | import java.util
|
24 |
| -import org.apache.hadoop.fs.{FileSystem, Path} |
| 22 | +import java.util.Properties |
| 23 | + |
| 24 | +import org.apache.spark.util.Utils |
| 25 | + |
| 26 | +import scala.collection.JavaConversions._ |
| 27 | + |
| 28 | +import org.apache.hadoop.conf.Configuration |
25 | 29 | import org.apache.hadoop.hive.serde2.{SerDeStats, AbstractSerDe}
|
26 |
| -import org.apache.hadoop.io.{NullWritable, Writable} |
| 30 | +import org.apache.hadoop.io.Writable |
27 | 31 | import org.apache.hadoop.hive.serde2.objectinspector.{ObjectInspectorFactory, ObjectInspector}
|
28 |
| -import java.util.Properties |
| 32 | + |
29 | 33 | import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory
|
30 |
| -import scala.collection.JavaConversions._ |
31 |
| -import java.io.{DataOutput, DataInput} |
32 | 34 | import org.apache.hadoop.hive.ql.udf.generic.GenericUDF
|
33 | 35 | import org.apache.hadoop.hive.ql.udf.generic.GenericUDF.DeferredObject
|
34 | 36 |
|
| 37 | +import org.apache.spark.sql.Row |
| 38 | +import org.apache.spark.sql.hive.test.TestHive |
| 39 | +import org.apache.spark.sql.hive.test.TestHive._ |
| 40 | + |
| 41 | +case class Fields(f1: Int, f2: Int, f3: Int, f4: Int, f5: Int) |
| 42 | + |
35 | 43 | /**
|
36 | 44 | * A test suite for Hive custom UDFs.
|
37 | 45 | */
|
38 | 46 | class HiveUdfSuite extends HiveComparisonTest {
|
39 | 47 |
|
40 |
| - TestHive.sql( |
41 |
| - """ |
| 48 | + test("spark sql udf test that returns a struct") { |
| 49 | + registerFunction("getStruct", (_: Int) => Fields(1, 2, 3, 4, 5)) |
| 50 | + assert(sql( |
| 51 | + """ |
| 52 | + |SELECT getStruct(1).f1, |
| 53 | + | getStruct(1).f2, |
| 54 | + | getStruct(1).f3, |
| 55 | + | getStruct(1).f4, |
| 56 | + | getStruct(1).f5 FROM src LIMIT 1 |
| 57 | + """.stripMargin).first() === Row(1, 2, 3, 4, 5)) |
| 58 | + } |
| 59 | + |
| 60 | + test("hive struct udf") { |
| 61 | + sql( |
| 62 | + """ |
42 | 63 | |CREATE EXTERNAL TABLE hiveUdfTestTable (
|
43 | 64 | | pair STRUCT<id: INT, value: INT>
|
44 | 65 | |)
|
45 | 66 | |PARTITIONED BY (partition STRING)
|
46 | 67 | |ROW FORMAT SERDE '%s'
|
47 | 68 | |STORED AS SEQUENCEFILE
|
48 |
| - """.stripMargin.format(classOf[PairSerDe].getName) |
49 |
| - ) |
50 |
| - |
51 |
| - TestHive.sql( |
52 |
| - "ALTER TABLE hiveUdfTestTable ADD IF NOT EXISTS PARTITION(partition='testUdf') LOCATION '%s'" |
53 |
| - .format(this.getClass.getClassLoader.getResource("data/files/testUdf").getFile) |
54 |
| - ) |
55 |
| - |
56 |
| - TestHive.sql("CREATE TEMPORARY FUNCTION testUdf AS '%s'".format(classOf[PairUdf].getName)) |
57 |
| - |
58 |
| - TestHive.sql("SELECT testUdf(pair) FROM hiveUdfTestTable") |
59 |
| - |
60 |
| - TestHive.sql("DROP TEMPORARY FUNCTION IF EXISTS testUdf") |
| 69 | + """. |
| 70 | + stripMargin.format(classOf[PairSerDe].getName)) |
| 71 | + |
| 72 | + val location = Utils.getSparkClassLoader.getResource("data/files/testUdf").getFile |
| 73 | + sql(s""" |
| 74 | + ALTER TABLE hiveUdfTestTable |
| 75 | + ADD IF NOT EXISTS PARTITION(partition='testUdf') |
| 76 | + LOCATION '$location'""") |
| 77 | + |
| 78 | + sql(s"CREATE TEMPORARY FUNCTION testUdf AS '${classOf[PairUdf].getName}'") |
| 79 | + sql("SELECT testUdf(pair) FROM hiveUdfTestTable") |
| 80 | + sql("DROP TEMPORARY FUNCTION IF EXISTS testUdf") |
| 81 | + } |
61 | 82 | }
|
62 | 83 |
|
63 | 84 | class TestPair(x: Int, y: Int) extends Writable with Serializable {
|
|
0 commit comments