[SPARK-19923][SQL] Remove unnecessary type conversions per call in Hive

[maropu]

What changes were proposed in this pull request?

This pr removed unnecessary type conversions per call in Hive: https://github.com/apache/spark/blob/master/sql/hive/src/main/scala/org/apache/spark/sql/hive/hiveUDFs.scala#L116

How was this patch tested?

Existing tests

[maropu]

I'm not sure this makes sense, so could you check? cc: @hvanhovell

[maropu]

I quickly did check performance changes;

public class TestGenericUDF extends GenericUDF {
  @Override
  public ObjectInspector initialize(ObjectInspector[] objectInspectors) throws UDFArgumentException {
      return PrimitiveObjectInspectorFactory.javaLongObjectInspector;
  }

  @Override
  public Object evaluate(DeferredObject[] args) throws HiveException {
      final long a1 = PrimitiveObjectInspectorFactory.javaLongObjectInspector.get(args[0].get());
      final long a2 = PrimitiveObjectInspectorFactory.javaLongObjectInspector.get(args[1].get());
      final long a3 = PrimitiveObjectInspectorFactory.javaLongObjectInspector.get(args[2].get());
      final long a4 = PrimitiveObjectInspectorFactory.javaLongObjectInspector.get(args[3].get());
      return a1 + a2 + a3 + a4;
  }
}

$./bin/spark-shell --master local[1]  --conf spark.sql.shuffle.partitions=1 -v

scala> sql("CREATE TEMPORARY FUNCTION testUdf AS 'hivemall.ftvec.TestGenericUDF'")
scala> :paste
def timer[R](block: => R): R = {
  val t0 = System.nanoTime()
  val result = block
  val t1 = System.nanoTime()
  println("Elapsed time: " + ((t1 - t0 + 0.0) / 1000000000.0)+ "s")
  result
}

scala> spark.range(3000000).createOrReplaceTempView("t")
scala> timer { sql("SELECT testUdf(id, id, id, id) FROM t").queryExecution.executedPlan.execute().foreach(x => {}) }

# performance w/o this patch
Elapsed time: 1.901269167s

# performance w/ this patch
Elapsed time: 0.492860666s

[SparkQA]

Test build #74396 has finished for PR 17264 at commit 6468fde.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[dongjoon-hyun]

Hi, @maropu .
This seems to be improvement, too. Could you add some description about this?
The current description only mentions hiveUDFs.scala.

[maropu]

Added the description to explain wrappers: https://github.com/apache/spark/pull/17264/files#diff-01999ccbf13e95a0ea2d223f69d8ae23R201

[HyukjinKwon]

(my personal taste that might be worth considering..

private[this] val wrappers = dataSchema
  .zip(structOI.getAllStructFieldRefs.asScala.toSeq)
  .map { case (f, i) => wrapperFor(i.getFieldObjectInspector, f.dataType) }

or maybe

private[this] val wrappers = 
  structOI.getAllStructFieldRefs.asScala.toSeq.zip(dataSchema).map { case (ref, field) =>
    wrapperFor(ref.getFieldObjectInspector, field.dataType)
  }

which resembles

spark/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveInspectors.scala

Lines 364 to 366 in 226d388

	val wrappers = ssoi.getAllStructFieldRefs.asScala.zip(structType).map {
	case (ref, tpe) => wrapperFor(ref.getFieldObjectInspector, tpe.dataType)
	}

)

[maropu]

Thanks for valuable comments! I'll update soon.

[hvanhovell]

@maropu this looks like a nice improvement. It is an idea to check all uses of wrap, and see if we can replace them by wrapperFor?

[maropu]

@hvanhovell I checked all the related code and then I think this pr includes all the place we could replace wrap with wrapperFor.

[SparkQA]

Test build #74537 has finished for PR 17264 at commit efd3628.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[hvanhovell]

LGTM merging to master. Thanks!

[rxin]

In the future can we put the perf result in PR descriptions?

[maropu]

@rxin okay!