Skip to content
Closed
Show file tree
Hide file tree
Changes from 3 commits
Commits
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
Original file line number Diff line number Diff line change
@@ -0,0 +1,118 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

package org.apache.spark.sql.hive.thriftserver

import java.sql.DatabaseMetaData
import java.util.UUID

import scala.collection.JavaConverters.seqAsJavaListConverter

import org.apache.hadoop.hive.ql.security.authorization.plugin.{HiveOperationType, HivePrivilegeObjectUtils}
import org.apache.hive.service.cli._
import org.apache.hive.service.cli.operation.GetFunctionsOperation
import org.apache.hive.service.cli.session.HiveSession
import org.apache.thrift.TException

import org.apache.spark.internal.Logging
import org.apache.spark.sql.SQLContext
import org.apache.spark.util.{Utils => SparkUtils}

/**
* Spark's own GetTablesOperation
Comment thread
wangyum marked this conversation as resolved.
Outdated
*
* @param sqlContext SQLContext to use
* @param parentSession a HiveSession from SessionManager
* @param catalogName catalog name. null if not applicable
* @param schemaName database name, null or a concrete database name
* @param functionName function name pattern
*/
private[hive] class SparkGetFunctionsOperation(
sqlContext: SQLContext,
parentSession: HiveSession,
catalogName: String,
schemaName: String,
functionName: String)
extends GetFunctionsOperation(parentSession, catalogName, schemaName, functionName)
with Logging {
Comment thread
wangyum marked this conversation as resolved.
Outdated

override def runInternal(): Unit = {
val statementId = UUID.randomUUID().toString
// Do not change cmdStr. It's used for Hive auditing and authorization.
val cmdStr = s"catalog : $catalogName, schemaPattern : $schemaName"
val logMsg = s"Listing functions '$cmdStr'"
Comment thread
wangyum marked this conversation as resolved.
Outdated
logInfo(s"$logMsg with $statementId")
setState(OperationState.RUNNING)
// Always use the latest class loader provided by executionHive's state.
val executionHiveClassLoader = sqlContext.sharedState.jarClassLoader
Thread.currentThread().setContextClassLoader(executionHiveClassLoader)

val catalog = sqlContext.sessionState.catalog

if (isAuthV2Enabled) {
// get databases for schema pattern
val schemaPattern = convertSchemaPattern(schemaName)
var matchingDbs: Seq[String] = null
try {
matchingDbs = catalog.listDatabases(schemaPattern)
} catch {
case e: TException =>
setState(OperationState.ERROR)
HiveThriftServer2.listener.onStatementError(
statementId, e.getMessage, SparkUtils.exceptionString(e))
throw new HiveSQLException(e)
}
// authorize this call on the schema objects
val privObjs =
HivePrivilegeObjectUtils.getHivePrivDbObjects(seqAsJavaListConverter(matchingDbs).asJava)
authorizeMetaGets(HiveOperationType.GET_FUNCTIONS, privObjs, cmdStr)
}

HiveThriftServer2.listener.onStatementStart(
statementId,
parentSession.getSessionHandle.getSessionId.toString,
logMsg,
statementId,
parentSession.getUsername)

try {

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.

This looks correct now: retrieve all functions available for all matching schemas.

val functionPattern = CLIServiceUtils.patternToRegex(functionName)
if ((null == catalogName || "".equals(catalogName))
&& (null == schemaName || "".equals(schemaName))) {
catalog.listFunctions(catalog.getCurrentDatabase, functionPattern).foreach {

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Actually, I'm confused about this code. Maybe it should be:

val functionPattern = CLIServiceUtils.patternToRegex(functionName)
matchingDbs.foreach { schema =>
  catalog.listFunctions(catalog.getCurrentDatabase, functionPattern).foreach {
    case (functionIdentifier, _) =>
      val rowData = Array[AnyRef](
        DEFAULT_HIVE_CATALOG, // FUNCTION_CAT
        schema, // FUNCTION_SCHEM
        functionIdentifier.funcName, // FUNCTION_NAME
        "", // REMARKS
        DatabaseMetaData.functionResultUnknown.asInstanceOf[AnyRef], // FUNCTION_TYPE
        "") // SPECIFIC_NAME
      rowSet.addRow(rowData);
}

But it's the logic of Hive: https://github.com/apache/hive/blob/rel/release-3.1.1/service/src/java/org/apache/hive/service/cli/operation/GetFunctionsOperation.java#L101-L119

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.

I think we should cover the case of functions that don't have a schema (null) which is basically what Hive's implementation seems to do, as well as the functions associated with a given schema which is what your code snippet above seems to do. Could you combine both?

https://docs.microsoft.com/en-us/sql/connect/jdbc/reference/getfunctions-method-sqlserverdatabasemetadata?view=sql-server-2017

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

case (functionIdentifier, _) =>
val rowData = Array[AnyRef](
null, // FUNCTION_CAT
Comment thread
wangyum marked this conversation as resolved.
Outdated
null, // FUNCTION_SCHEM
functionIdentifier.funcName, // FUNCTION_NAME
"", // REMARKS

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.

I wonder if we can get the function usage somehow from catalog ...

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Done.

catalog.lookupFunctionInfo(funcIdentifier).getUsage

DatabaseMetaData.functionResultUnknown.asInstanceOf[AnyRef], // FUNCTION_TYPE

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

We do not support FUNCTION_TYPE now. Set it to Unknown:

    // java.sql.DatabaseMetaData

    /**
     * Indicates that it is not known whether the function returns
     * a result or a table.
     * <P>
     * A possible value for column <code>FUNCTION_TYPE</code> in the
     * <code>ResultSet</code> object returned by the method
     * <code>getFunctions</code>.
     * @since 1.6
     */
    int functionResultUnknown   = 0;

    /**
     * Indicates that the function  does not return a table.
     * <P>
     * A possible value for column <code>FUNCTION_TYPE</code> in the
     * <code>ResultSet</code> object returned by the method
     * <code>getFunctions</code>.
     * @since 1.6
     */
    int functionNoTable         = 1;

    /**
     * Indicates that the function  returns a table.
     * <P>
     * A possible value for column <code>FUNCTION_TYPE</code> in the
     * <code>ResultSet</code> object returned by the method
     * <code>getFunctions</code>.
     * @since 1.6
     */
    int functionReturnsTable    = 2;

"")

@bogdanghit bogdanghit Jul 30, 2019

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.

This is the function class name which I think we can get through a catalog.getFunction(funcIdentifier).className call.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Done.

catalog.lookupFunctionInfo(funcIdentifier).getClassName

rowSet.addRow(rowData);
}
}
setState(OperationState.FINISHED)
} catch {
case e: HiveSQLException =>
setState(OperationState.ERROR)
HiveThriftServer2.listener.onStatementError(
statementId, e.getMessage, SparkUtils.exceptionString(e))
throw e
}

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.

Shouldn't we handle other exceptions too?

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Maybe it should be the same as SparkExecuteStatementOperation:

// Actually do need to catch Throwable as some failures don't inherit from Exception and
// HiveServer will silently swallow them.
case e: Throwable =>
val currentState = getStatus().getState()
logError(s"Error executing query, currentState $currentState, ", e)
setState(OperationState.ERROR)
HiveThriftServer2.listener.onStatementError(
statementId, e.getMessage, SparkUtils.exceptionString(e))
throw new HiveSQLException(e.toString)

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.

Or the same as GetTablesOperation and GetSchemasOperation for the sake of consistency?

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

+1 for same as GetTablesOperation and GetSchemasOperation.

HiveThriftServer2.listener.onStatementFinish(statementId)

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.

We also need the onStatementClosed handler like in the other ops.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Done

}
}
Original file line number Diff line number Diff line change
Expand Up @@ -119,6 +119,21 @@ private[thriftserver] class SparkSQLOperationManager()
operation
}

override def newGetFunctionsOperation(
parentSession: HiveSession,
catalogName: String,
schemaName: String,
functionName: String): GetFunctionsOperation = synchronized {
val sqlContext = sessionToContexts.get(parentSession.getSessionHandle)
require(sqlContext != null, s"Session handle: ${parentSession.getSessionHandle} has not been" +
" initialized or had already closed.")
val operation = new SparkGetFunctionsOperation(sqlContext, parentSession,
catalogName, schemaName, functionName)
handleToOperation.put(operation.getHandle, operation)
logDebug(s"Created GetFunctionsOperation with session=$parentSession.")
operation
}

def setConfMap(conf: SQLConf, confMap: java.util.Map[String, String]): Unit = {
val iterator = confMap.entrySet().iterator()
while (iterator.hasNext) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@

package org.apache.spark.sql.hive.thriftserver

import java.sql.ResultSet
import java.sql.{DatabaseMetaData, ResultSet}

class SparkMetadataOperationSuite extends HiveThriftJdbcTest {

Expand Down Expand Up @@ -182,4 +182,26 @@ class SparkMetadataOperationSuite extends HiveThriftJdbcTest {
checkResult(metaData.getTableTypes, Seq("TABLE", "VIEW"))
}
}

@bogdanghit bogdanghit Jul 31, 2019

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.

Can you add some tests for the usage and the function class name? @wangyum

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

I have added some tests:

ssert(rs.getString("REMARKS").startsWith(s"${functionName(i)}("))
assert(rs.getString("SPECIFIC_NAME").startsWith("org.apache.spark.sql.catalyst"))

Do you think we need to assert more details?

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.

Would be nice to run a DESCRIBE function statement and then compare the results.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

test("Spark's own GetFunctionsOperation(SparkGetFunctionsOperation)") {
def checkResult(rs: ResultSet, functionName: Seq[String]): Unit = {
for (i <- functionName.indices) {
assert(rs.next())
assert(rs.getString("FUNCTION_NAME") === functionName(i))
assert(rs.getInt("FUNCTION_TYPE") === DatabaseMetaData.functionResultUnknown)
}
// Make sure there are no more elements
assert(!rs.next())
}

withJdbcStatement() { statement =>
val metaData = statement.getConnection.getMetaData
// Hive does not have an overlay function, we use overlay to test.
checkResult(metaData.getFunctions(null, null, "overlay"), Seq("overlay"))
checkResult(metaData.getFunctions(null, null, "overla*"), Seq("overlay"))
checkResult(metaData.getFunctions(null, "", "overla*"), Seq("overlay"))
checkResult(metaData.getFunctions(null, null, "does-not-exist*"), Seq.empty)
checkResult(metaData.getFunctions(null, "default", "overlay"), Seq.empty)
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -63,7 +63,7 @@ public class GetFunctionsOperation extends MetadataOperation {
private final String schemaName;
private final String functionName;

private final RowSet rowSet;
protected final RowSet rowSet;

public GetFunctionsOperation(HiveSession parentSession,
String catalogName, String schemaName, String functionName) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -63,7 +63,7 @@ public class GetFunctionsOperation extends MetadataOperation {
private final String schemaName;
private final String functionName;

private final RowSet rowSet;
protected final RowSet rowSet;

public GetFunctionsOperation(HiveSession parentSession,
String catalogName, String schemaName, String functionName) {
Expand Down