Skip to content

[SPARK-20033][SQL] support hive permanent function #17362

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
cenyuhai wants to merge 4 commits into from
Closed

[SPARK-20033][SQL] support hive permanent function #17362

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
703d23e
support hive permanent function
cenyuhai Mar 20, 2017
a668bb5
Don't remove file when testing
cenyuhai Mar 20, 2017
9325a7c
fix compile error
cenyuhai Mar 20, 2017
92c077d
check resouce scheme
cenyuhai Mar 20, 2017
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
40 changes: 38 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 @@ -17,27 +17,34 @@

package org.apache.spark.sql.hive

import java.io.File
import java.net.URI

import scala.util.{Failure, Success, Try}
import scala.util.control.NonFatal

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.Path
import org.apache.hadoop.hive.ql.exec.{UDAF, UDF}
import org.apache.hadoop.hive.ql.exec.{FunctionRegistry => HiveFunctionRegistry}
import org.apache.hadoop.hive.ql.session.SessionState
import org.apache.hadoop.hive.ql.session.SessionState.ResourceType
import org.apache.hadoop.hive.ql.udf.generic.{AbstractGenericUDAFResolver, GenericUDF, GenericUDTF}
import org.apache.hadoop.util.Shell

import org.apache.spark.sql.{AnalysisException, SparkSession}
import org.apache.spark.sql.catalyst.{FunctionIdentifier, TableIdentifier}
import org.apache.spark.sql.catalyst.analysis.FunctionRegistry
import org.apache.spark.sql.catalyst.analysis.FunctionRegistry.FunctionBuilder
import org.apache.spark.sql.catalyst.catalog.{FunctionResourceLoader, GlobalTempViewManager, SessionCatalog}
import org.apache.spark.sql.catalyst.catalog._
import org.apache.spark.sql.catalyst.expressions.{Cast, Expression, ExpressionInfo}
import org.apache.spark.sql.catalyst.parser.ParserInterface
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.hive.HiveShim.HiveFunctionWrapper
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{DecimalType, DoubleType}
import org.apache.spark.util.Utils
import org.apache.spark.util.{ShutdownHookManager, Utils}


private[sql] class HiveSessionCatalog(
Expand Down Expand Up @@ -135,6 +142,35 @@ private[sql] class HiveSessionCatalog(
}
}

override def loadFunctionResources(resources: Seq[FunctionResource]): Unit = {
logDebug("loading hive permanent function resources")
resources.foreach { resource =>
val resourceType = resource.resourceType match {
case JarResource =>
ResourceType.JAR
case FileResource =>
ResourceType.FILE
case ArchiveResource =>
ResourceType.ARCHIVE
}
val uri = if (!Shell.WINDOWS) {
new URI(resource.uri)
}
else {
new Path(resource.uri).toUri
}
val scheme = if (uri.getScheme == null) null else uri.getScheme.toLowerCase
if (scheme == null || scheme == "file") {
functionResourceLoader.loadResource(resource)
} else {
val sessionState = SessionState.get()
val localPath = sessionState.add_resource(resourceType, resource.uri)
Copy link
Member

@gatorsmile gatorsmile Mar 23, 2017

Choose a reason for hiding this comment

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

Are you trying to create a Hive UDF via Spark, then call it through Hive?

Copy link
Contributor Author

@cenyuhai cenyuhai Mar 24, 2017

Choose a reason for hiding this comment

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

no, I just use sessionState.add_resource to download the resouces

ShutdownHookManager.registerShutdownDeleteDir(new File(localPath).getParentFile)
functionResourceLoader.loadResource(FunctionResource(resource.resourceType, localPath))
}
}
}

override def lookupFunction(name: FunctionIdentifier, children: Seq[Expression]): Expression = {
try {
lookupFunction0(name, children)
Expand Down