Skip to content

[SPARK-33692][SQL] View should use captured catalog and namespace to lookup function #30662

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
linhongliu-db wants to merge 10 commits into from
Closed

[SPARK-33692][SQL] View should use captured catalog and namespace to lookup function #30662

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
f3c2fd6
[SPARK-33692][SQL] View should use captured catalog and namespace to …
linhongliu-db Dec 8, 2020
c104cfe
test fix
linhongliu-db Dec 8, 2020
00e32d6
code clean
linhongliu-db Dec 8, 2020
ef7c6b2
Update UDFSuite.scala
linhongliu-db Dec 9, 2020
d87f249
fix test cases in temp view
linhongliu-db Dec 9, 2020
11523f4
remove code
linhongliu-db Dec 9, 2020
a8c7722
another way
linhongliu-db Dec 10, 2020
ab440e3
add test cases
linhongliu-db Dec 10, 2020
b30272c
code clean
linhongliu-db Dec 10, 2020
632f3d6
address comment
linhongliu-db Dec 10, 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
10 changes: 6 additions & 4 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -114,7 +114,8 @@ case class AnalysisContext(
nestedViewDepth: Int = 0,
maxNestedViewDepth: Int = -1,
relationCache: mutable.Map[Seq[String], LogicalPlan] = mutable.Map.empty,
referredTempViewNames: Seq[Seq[String]] = Seq.empty)
referredTempViewNames: Seq[Seq[String]] = Seq.empty,
isTempView: Boolean = false)

@cloud-fan cloud-fan Dec 9, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

then shall we just use referredTempFuncNames?

@linhongliu-db linhongliu-db Dec 9, 2020 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

if we change the type to referredTempFuncNames: Option[Seq[String]], then it can replace isTempView. Because Seq.empty can be permanent view or temp view refers non temp functions.

@cloud-fan cloud-fan Dec 9, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

yea let's do it then. We can use isTempView: Boolean = false in the backport PR of 3.0/2.4.

@linhongliu-db linhongliu-db Dec 10, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

you are correct in the first comment. we can just use referredTempFunctionNames. But this approach will fix both permanent and temp view problems, which I was planning to fix them separately. Anyway, I switch to use referredTempFunctionNames in the latest commit.

BTW, in 3.0/2.4, temp view stores analyzed plan, and won't be wrapped in View, so we don't need the isTempView flag.


object AnalysisContext {
private val value = new ThreadLocal[AnalysisContext]() {
Expand All @@ -126,7 +127,7 @@ object AnalysisContext {

private def set(context: AnalysisContext): Unit = value.set(context)

def withAnalysisContext[A](viewDesc: CatalogTable)(f: => A): A = {
def withAnalysisContext[A](viewDesc: CatalogTable, isTempView: Boolean)(f: => A): A = {
val originContext = value.get()
val maxNestedViewDepth = if (originContext.maxNestedViewDepth == -1) {
// Here we start to resolve views, get `maxNestedViewDepth` from configs.
Expand All @@ -139,7 +140,8 @@ object AnalysisContext {
originContext.nestedViewDepth + 1,
maxNestedViewDepth,
originContext.relationCache,
viewDesc.viewReferredTempViewNames)
viewDesc.viewReferredTempViewNames,
isTempView)
set(context)
try f finally { set(originContext) }
}
Expand Down Expand Up @@ -1048,7 +1050,7 @@ class Analyzer(override val catalogManager: CatalogManager)
// operator.
case view @ View(desc, isTempView, _, child) if !child.resolved =>
// Resolve all the UnresolvedRelations and Views in the child.
val newChild = AnalysisContext.withAnalysisContext(desc) {
val newChild = AnalysisContext.withAnalysisContext(desc, isTempView) {
val nestedViewDepth = AnalysisContext.get.nestedViewDepth
val maxNestedViewDepth = AnalysisContext.get.maxNestedViewDepth
if (nestedViewDepth > maxNestedViewDepth) {
Expand Down
23 changes: 20 additions & 3 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1488,12 +1488,29 @@ class SessionCatalog(
// We probably shouldn't use a single FunctionRegistry to register all three kinds of functions
// (built-in, temp, and external).
if (name.database.isEmpty && functionRegistry.functionExists(name)) {
// This function has been already loaded into the function registry.
return functionRegistry.lookupFunction(name, children)
val isResolvingPermanentView =
AnalysisContext.get.catalogAndNamespace.nonEmpty && !AnalysisContext.get.isTempView
// We lookup function without database in two cases:
// 1. the function is not a temporary function

@cloud-fan cloud-fan Dec 9, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

the function is built-in

@linhongliu-db linhongliu-db Dec 9, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

there are also external functions, in isTemporaryFunction:

    name.database.isEmpty &&
      functionRegistry.functionExists(name) &&
      !FunctionRegistry.builtin.functionExists(name) &&
      !hiveFunctions.contains(name.funcName.toLowerCase(Locale.ROOT))

// 2. the function is a temporary function but we are not resolving permanent view
if (!isTemporaryFunction(name) || !isResolvingPermanentView) {
// This function has been already loaded into the function registry.
return functionRegistry.lookupFunction(name, children)
}
}

// Get the database from AnalysisContext if it's defined, otherwise, use current database
val currentDatabase = AnalysisContext.get.catalogAndNamespace match {
case Seq() => getCurrentDatabase
case Seq(_, db) => db
case Seq(catalog, namespace @ _*) =>
throw new AnalysisException(
s"V2 catalog does not support functions yet. " +
s"catalog: ${catalog}, namespace '${namespace.mkString("[", ".", "]")}'")

@cloud-fan cloud-fan Dec 10, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

nit: namespace.quoted. We need to import CatalogV2Implicits._ first.

}

// If the name itself is not qualified, add the current database to it.
val database = formatDatabaseName(name.database.getOrElse(getCurrentDatabase))
val database = formatDatabaseName(name.database.getOrElse(currentDatabase))
val qualifiedName = name.copy(database = Some(database))

if (functionRegistry.functionExists(qualifiedName)) {
Expand Down
28 changes: 28 additions & 0 deletions sql/hive/src/test/scala/org/apache/spark/sql/hive/UDFSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -209,4 +209,32 @@ class UDFSuite
assert(e.getMessage.contains("Can not get an evaluator of the empty UDAF"))
}
}

test("SPARK-33692: permanent view should use captured catalog and namespace for function") {
val upperFuncClass =
classOf[org.apache.hadoop.hive.ql.udf.generic.GenericUDFUpper].getCanonicalName
val lowerFuncClass =
classOf[org.apache.hadoop.hive.ql.udf.generic.GenericUDFLower].getCanonicalName
val functionName = "test_udf"
withTempDatabase { dbName =>
withUserDefinedFunction(
s"default.$functionName" -> false,
s"$dbName.$functionName" -> false,
functionName -> true) {
withView("default.v1") {
sql("USE DEFAULT")
sql(s"CREATE FUNCTION $functionName AS '$upperFuncClass'")
// create a view using a function in 'default' database
sql(s"CREATE VIEW v1 AS SELECT $functionName('TeSt_STr')")
sql(s"USE $dbName")
// create function in another database with the same function name
sql(s"CREATE FUNCTION $functionName AS '$lowerFuncClass'")
// create temporary function with the same function name
sql(s"CREATE TEMPORARY FUNCTION $functionName AS '$lowerFuncClass'")
// view v1 should still using function defined in `default` database
checkAnswer(sql("SELECT * FROM default.v1"), Seq(Row("TEST_STR")))
}
}
}
}
}