Skip to content

[SPARK-33704][SQL] Support latest version of initialize() in HiveGenericUDTF #30665

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
southernriver wants to merge 3 commits into from
Closed

[SPARK-33704][SQL] Support latest version of initialize() in HiveGenericUDTF #30665

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
e78421d
Suit the latest version of initialize(StructObjectInspector argOIs) f…
southernriver Jul 15, 2020
ea65856
Reset IllegalStateException of udtf_stack2 to be normal
Dec 8, 2020
0e3dca1
fix style
Dec 9, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Binary file added build/._scala-2.12.10
View file
Open in desktop
Binary file not shown.
Binary file added build/._zinc-0.3.15
View file
Open in desktop
Binary file not shown.
14 changes: 12 additions & 2 deletions sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveSessionCatalog.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,8 +91,18 @@ private[sql] class HiveSessionCatalog(
isUDAFBridgeRequired = true))
udfExpr.get.dataType // Force it to check input data types.
} else if (classOf[GenericUDTF].isAssignableFrom(clazz)) {
udfExpr = Some(HiveGenericUDTF(name, new HiveFunctionWrapper(clazz.getName), input))
udfExpr.get.asInstanceOf[HiveGenericUDTF].elementSchema // Force it to check data types.
try {
udfExpr = Some(HiveGenericUDTF(name, new HiveFunctionWrapper(clazz.getName), input))
// Force it to check input data types.
udfExpr.get.asInstanceOf[HiveGenericUDTF].elementSchema
} catch {
case exception: Exception =>
Copy link
Member

@sunchao sunchao Dec 30, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why do we need this fallback mechanism? can we just switch to the new API without the deprecated flag?

logInfo(s"HiveGenericUDTF initialize(ObjectInspector[] args) is deprecated, and" +
s" we will suit the latest version of initialize(StructObjectInspector argOIs).")
udfExpr = Some(HiveGenericUDTF(name, new HiveFunctionWrapper(clazz.getName),
input, false))
udfExpr.get.asInstanceOf[HiveGenericUDTF].elementSchema
Comment on lines +94 to +104
Copy link
Member

@wangyum wangyum Dec 9, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How about?

val funcWrapper = new HiveFunctionWrapper(clazz.getName)
try {
  udfExpr = Some(HiveGenericUDTF(name, funcWrapper, input, true))
  // Force it to check data types.
  udfExpr.get.asInstanceOf[HiveGenericUDTF].elementSchema
} catch {
  case e: IllegalStateException if e.getMessage.equals("Should not be called directly") =>
    logInfo("Fallback to use the non deprecated UDTF constructor.")
    udfExpr = Some(HiveGenericUDTF(name, funcWrapper, input, false))
    // Force it to check data types.
    udfExpr.get.asInstanceOf[HiveGenericUDTF].elementSchema
}

}
}
} catch {
case NonFatal(e) =>
Expand Down
17 changes: 15 additions & 2 deletions sql/hive/src/main/scala/org/apache/spark/sql/hive/hiveUDFs.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -198,7 +198,8 @@ private[hive] case class HiveGenericUDF(
private[hive] case class HiveGenericUDTF(
name: String,
funcWrapper: HiveFunctionWrapper,
children: Seq[Expression])
children: Seq[Expression],
deprecated: Boolean = true)
Copy link
Member

@wangyum wangyum Dec 9, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

deprecated: Boolean = true -> isDeprecatedConstructor: Boolean?

extends Generator with HiveInspectors with CodegenFallback with UserDefinedExpression {

@transient
Expand All @@ -212,7 +213,19 @@ private[hive] case class HiveGenericUDTF(
protected lazy val inputInspectors = children.map(toInspector)

@transient
protected lazy val outputInspector = function.initialize(inputInspectors.toArray)
protected lazy val inpuColNames = children.map(_ => "field_name").asJava

@transient
protected lazy val rowOI = ObjectInspectorFactory.getStandardStructObjectInspector(
inpuColNames, inputInspectors.asJava)

@transient
protected lazy val outputInspector =
if (deprecated) {
function.initialize(inputInspectors.toArray)
} else {
function.initialize(rowOI)
}
Comment on lines +223 to +228
Copy link
Member

@wangyum wangyum Dec 9, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How about?

protected lazy val outputInspector = {
  if (isDeprecatedConstructor) {
    function.initialize(inputInspectors.toArray)
  } else {
    val rowOI = ObjectInspectorFactory.getStandardStructObjectInspector(
      children.zipWithIndex.map(e => s"_col${e._2}").asJava, inputInspectors.asJava)
    function.initialize(rowOI)
  }
}


@transient
protected lazy val udtInput = new Array[AnyRef](children.length)
Expand Down
8 changes: 3 additions & 5 deletions sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLQuerySuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2132,11 +2132,9 @@ class SQLQuerySuite extends QueryTest with SQLTestUtils with TestHiveSingleton {
|AS 'org.apache.spark.sql.hive.execution.UDTFStack2'
|USING JAR '${hiveContext.getHiveFile("SPARK-21101-1.0.jar").toURI}'
""".stripMargin)
val e = intercept[org.apache.spark.sql.AnalysisException] {
sql("SELECT udtf_stack2(2, 'A', 10, date '2015-01-01', 'B', 20, date '2016-01-01')")
}
assert(
e.getMessage.contains("public StructObjectInspector initialize(ObjectInspector[] args)"))
val num =
sql("SELECT udtf_stack2(2, 'A', 10, date '2015-01-01', 'B', 20, date '2016-01-01')").count()
assert(num === 2)
Comment on lines +2135 to +2137
Copy link
Member

@wangyum wangyum Dec 9, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How about?

checkAnswer(
  sql("SELECT udtf_stack2(2, 'A', 10, date '2015-01-01', 'B', 20, date '2016-01-01')"),
  Seq(Row("A", 10, java.sql.Date.valueOf("2015-01-01")),
    Row("B", 20, java.sql.Date.valueOf("2016-01-01"))))

Copy link
Member

@wangyum wangyum Dec 9, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could you update test name?

}
}

Expand Down