forked from apache/incubator-livy
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[LIVY-622][LIVY-623][LIVY-624][LIVY-625][THRIFT] Support GetFunctions…
…, GetSchemas, GetTables, GetColumns in Livy thrift server ## What changes were proposed in this pull request? In this patch, we add the implementations of GetSchemas, GetFunctions, GetTables, and GetColumns in Livy Thrift server. https://issues.apache.org/jira/browse/LIVY-622 https://issues.apache.org/jira/browse/LIVY-623 https://issues.apache.org/jira/browse/LIVY-624 https://issues.apache.org/jira/browse/LIVY-625 ## How was this patch tested? Add new unit tests and integration test. Run them with existing tests. Author: yihengwang <[email protected]> Closes apache#194 from yiheng/fix_575.
- Loading branch information
Showing
22 changed files
with
1,395 additions
and
15 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
102 changes: 102 additions & 0 deletions
102
...er/server/src/main/scala/org/apache/livy/thriftserver/operation/GetColumnsOperation.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,102 @@ | ||
| /* | ||
| * 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.livy.thriftserver.operation | ||
|
|
||
| import org.apache.hive.service.cli.{HiveSQLException, OperationState, OperationType, SessionHandle} | ||
|
|
||
| import org.apache.livy.Logging | ||
| import org.apache.livy.thriftserver.types.{BasicDataType, Field, Schema} | ||
| import org.apache.livy.thriftserver.LivyThriftSessionManager | ||
| import org.apache.livy.thriftserver.session.{GetColumnsJob, GetFunctionsJob} | ||
|
|
||
| class GetColumnsOperation( | ||
| sessionHandle: SessionHandle, | ||
| catalogName: String, | ||
| schemaName: String, | ||
| tableName: String, | ||
| columnName: String, | ||
| sessionManager: LivyThriftSessionManager) | ||
| extends SparkCatalogOperation( | ||
| sessionHandle, OperationType.GET_COLUMNS, sessionManager) with Logging { | ||
|
|
||
| @throws(classOf[HiveSQLException]) | ||
| override protected def runInternal(): Unit = { | ||
| setState(OperationState.RUNNING) | ||
| try { | ||
| rscClient.submit(new GetColumnsJob( | ||
| convertSchemaPattern(schemaName), | ||
| convertIdentifierPattern(tableName, datanucleusFormat = true), | ||
| Option(columnName).map { convertIdentifierPattern(_, datanucleusFormat = false) }.orNull, | ||
| sessionId, | ||
| jobId | ||
| )).get() | ||
|
|
||
| setState(OperationState.FINISHED) | ||
| } catch { | ||
| case e: Throwable => | ||
| error("Remote job meet an exception: ", e) | ||
| setState(OperationState.ERROR) | ||
| throw new HiveSQLException(e) | ||
| } | ||
| } | ||
|
|
||
| @throws(classOf[HiveSQLException]) | ||
| override def getResultSetSchema: Schema = { | ||
| assertState(Seq(OperationState.FINISHED)) | ||
| GetColumnsOperation.SCHEMA | ||
| } | ||
| } | ||
|
|
||
| object GetColumnsOperation { | ||
| val SCHEMA = Schema( | ||
| Field("TABLE_CAT", BasicDataType("string"), "Catalog name. NULL if not applicable."), | ||
| Field("TABLE_SCHEM", BasicDataType("string"), "Schema name."), | ||
| Field("TABLE_NAME", BasicDataType("string"), "Table name."), | ||
| Field("COLUMN_NAME", BasicDataType("string"), "Column name"), | ||
| Field("DATA_TYPE", BasicDataType("integer"), "SQL type from java.sql.Types"), | ||
| Field("TYPE_NAME", BasicDataType("string"), | ||
| "Data source dependent type name, for a UDT the type name is fully qualified"), | ||
| Field("COLUMN_SIZE", BasicDataType("integer"), "Column size. For char or date types this is " + | ||
| "the maximum number of characters, for numeric or decimal types this is precision."), | ||
| Field("BUFFER_LENGTH", BasicDataType("byte"), "Unused"), | ||
| Field("DECIMAL_DIGITS", BasicDataType("integer"), "The number of fractional digits"), | ||
| Field("NUM_PREC_RADIX", BasicDataType("integer"), "Radix (typically either 10 or 2)"), | ||
| Field("NULLABLE", BasicDataType("integer"), "Is NULL allowed"), | ||
| Field("REMARKS", BasicDataType("string"), "Comment describing column (may be null)"), | ||
| Field("COLUMN_DEF", BasicDataType("string"), "Default value (may be null)"), | ||
| Field("SQL_DATA_TYPE", BasicDataType("integer"), "Unused"), | ||
| Field("SQL_DATETIME_SUB", BasicDataType("integer"), "Unused"), | ||
| Field("CHAR_OCTET_LENGTH", BasicDataType("integer"), "For char types the maximum number of " + | ||
| "bytes in the column"), | ||
| Field("ORDINAL_POSITION", BasicDataType("integer"), "Index of column in table (starting at 1)"), | ||
| Field("IS_NULLABLE", BasicDataType("string"), "\"NO\" means column definitely does not " + | ||
| "allow NULL values; \"YES\" means the column might allow NULL values. An empty string " + | ||
| "means nobody knows."), | ||
| Field("SCOPE_CATALOG", BasicDataType("string"), "Catalog of table that is the scope of a " + | ||
| "reference attribute (null if DATA_TYPE isn't REF)"), | ||
| Field("SCOPE_SCHEMA", BasicDataType("string"), "Schema of table that is the scope of a " + | ||
| "reference attribute (null if the DATA_TYPE isn't REF)"), | ||
| Field("SCOPE_TABLE", BasicDataType("string"), "Table name that this the scope of a " + | ||
| "reference attribure (null if the DATA_TYPE isn't REF)"), | ||
| Field("SOURCE_DATA_TYPE", BasicDataType("short"), "Source type of a distinct type or " + | ||
| "user-generated Ref type, SQL type from java.sql.Types (null if DATA_TYPE isn't " + | ||
| "DISTINCT or user-generated REF)"), | ||
| Field("IS_AUTO_INCREMENT", BasicDataType("string"), "Indicates whether this column is " + | ||
| "auto incremented.") | ||
| ) | ||
| } |
94 changes: 94 additions & 0 deletions
94
.../server/src/main/scala/org/apache/livy/thriftserver/operation/GetFunctionsOperation.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,94 @@ | ||
| /* | ||
| * 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.livy.thriftserver.operation | ||
|
|
||
| import org.apache.hive.service.cli.{HiveSQLException, OperationState, OperationType, SessionHandle} | ||
|
|
||
| import org.apache.livy.Logging | ||
| import org.apache.livy.thriftserver.session.GetFunctionsJob | ||
| import org.apache.livy.thriftserver.types.{BasicDataType, Field, Schema} | ||
| import org.apache.livy.thriftserver.LivyThriftSessionManager | ||
|
|
||
| class GetFunctionsOperation( | ||
| sessionHandle: SessionHandle, | ||
| catalogName: String, | ||
| schemaName: String, | ||
| functionName: String, | ||
| sessionManager: LivyThriftSessionManager) | ||
| extends SparkCatalogOperation( | ||
| sessionHandle, OperationType.GET_FUNCTIONS, sessionManager) with Logging { | ||
|
|
||
| @throws(classOf[HiveSQLException]) | ||
| override protected def runInternal(): Unit = { | ||
| setState(OperationState.RUNNING) | ||
| try { | ||
| rscClient.submit(new GetFunctionsJob( | ||
| convertSchemaPattern(schemaName), | ||
| convertFunctionName(functionName), | ||
| sessionId, | ||
| jobId | ||
| )).get() | ||
|
|
||
| setState(OperationState.FINISHED) | ||
| } catch { | ||
| case e: Throwable => | ||
| error("Remote job meet an exception: ", e) | ||
| setState(OperationState.ERROR) | ||
| throw new HiveSQLException(e) | ||
| } | ||
| } | ||
|
|
||
| @throws(classOf[HiveSQLException]) | ||
| override def getResultSetSchema: Schema = { | ||
| assertState(Seq(OperationState.FINISHED)) | ||
| GetFunctionsOperation.SCHEMA | ||
| } | ||
|
|
||
| private def convertFunctionName(name: String): String = { | ||
| if (name == null) { | ||
| ".*" | ||
| } else { | ||
| var escape = false | ||
| name.flatMap { | ||
| case c if escape => | ||
| if (c != '\\') escape = false | ||
| c.toString | ||
| case '\\' => | ||
| escape = true | ||
| "" | ||
| case '%' => ".*" | ||
| case '_' => "." | ||
| case c => Character.toLowerCase(c).toString | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| object GetFunctionsOperation { | ||
| val SCHEMA = Schema( | ||
| Field("FUNCTION_CAT", BasicDataType("string"), "Function catalog (may be null)"), | ||
| Field("FUNCTION_SCHEM", BasicDataType("string"), "Function schema (may be null)"), | ||
| Field("FUNCTION_NAME", BasicDataType("string"), | ||
| "Function name. This is the name used to invoke the function"), | ||
| Field("REMARKS", BasicDataType("string"), "Explanatory comment on the function"), | ||
| Field("FUNCTION_TYPE", BasicDataType("integer"), | ||
| "Kind of function."), | ||
| Field("SPECIFIC_NAME", BasicDataType("string"), | ||
| "The name which uniquely identifies this function within its schema") | ||
| ) | ||
| } |
Oops, something went wrong.