From 90f016727fd3a83a002088f02ebeba3b10349476 Mon Sep 17 00:00:00 2001 From: Adrian Ionescu Date: Wed, 22 Feb 2017 18:31:05 +0100 Subject: [PATCH] [SC-5840][DIRECTORYCOMMIT] Move Vacuum-related code to com.databricks.sql ## What changes were proposed in this pull request? This patch moves all Vacuum-related code from `org.apache.spark` to `com.databricks.sql` as part of the general task of clearly separating Edge issues in order to reduce merge conflicts with OSS. `AclCommandParser` is renamed to a more general `DatabricksSqlParser`, to be used for all DB-specific syntax and is moved to a new package called `com.databricks.sql.parser`. `VacuumTableCommand` is moved from `org.apache.spark.sql.execution.command` to `com.databricks.sql.transaction`. ## How was this patch tested? Tests in project `spark-sql` pass. Author: Adrian Ionescu Closes #242 from adrian-ionescu/SC-5840. --- project/SparkBuild.scala | 2 +- .../spark/sql/catalyst/parser/SqlBase.g4 | 5 -- .../DatabricksSqlBase.g4} | 68 ++++++++++++++++--- .../databricks/sql/acl/AclExtensions.scala | 3 +- .../databricks/sql/acl/CheckPermissions.scala | 1 + .../DatabricksSqlCommandBuilder.scala} | 39 ++++++++--- .../DatabricksSqlParser.scala} | 21 +++--- .../sql/transaction}/VacuumTableCommand.scala | 3 +- .../org/apache/spark/sql/SparkSession.scala | 4 ++ .../spark/sql/execution/SparkSqlParser.scala | 15 ---- .../sql/acl/CheckPermissionRuleSuite.scala | 1 + .../DatabricksSqlCommandParserSuite.scala} | 33 ++++++++- .../DatabricksAtomicCommitProtocolSuite.scala | 17 ++++- .../spark/sql/test/TestSQLContext.scala | 18 +++-- 14 files changed, 168 insertions(+), 62 deletions(-) rename sql/core/src/main/antlr4/com/databricks/sql/{acl/AclCommandBase.g4 => parser/DatabricksSqlBase.g4} (61%) rename sql/core/src/main/scala/com/databricks/sql/{acl/AstCommandBuilder.scala => parser/DatabricksSqlCommandBuilder.scala} (82%) rename sql/core/src/main/scala/com/databricks/sql/{acl/AclCommandParser.scala => parser/DatabricksSqlParser.scala} (85%) rename sql/core/src/main/scala/{org/apache/spark/sql/execution/command => com/databricks/sql/transaction}/VacuumTableCommand.scala (96%) rename sql/core/src/test/scala/com/databricks/sql/{acl/AclCommandParseSuite.scala => parser/DatabricksSqlCommandParserSuite.scala} (88%) diff --git a/project/SparkBuild.scala b/project/SparkBuild.scala index 60e81716e385a..7d27b14812c7f 100644 --- a/project/SparkBuild.scala +++ b/project/SparkBuild.scala @@ -548,7 +548,7 @@ object Catalyst { object SQL { lazy val settings = antlr4Settings ++ Seq( - antlr4PackageName in Antlr4 := Some("com.databricks.sql.acl"), + antlr4PackageName in Antlr4 := Some("com.databricks.sql.parser"), antlr4GenListener in Antlr4 := true, antlr4GenVisitor in Antlr4 := true, initialCommands in console := diff --git a/sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4 b/sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4 index 398e3f4e4442c..8d3f73271874d 100644 --- a/sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4 +++ b/sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4 @@ -150,7 +150,6 @@ statement | SET ROLE .*? #failNativeCommand | SET .*? #setConfiguration | RESET #resetConfiguration - | VACUUM (path=STRING | tableIdentifier) (RETAIN number HOURS)? #vacuumTable | unsupportedHiveNativeCommands .*? #failNativeCommand ; @@ -702,7 +701,6 @@ nonReserved | AND | CASE | CAST | DISTINCT | DIV | ELSE | END | FUNCTION | INTERVAL | MACRO | OR | STRATIFY | THEN | UNBOUNDED | WHEN | DATABASE | SELECT | FROM | WHERE | HAVING | TO | TABLE | WITH | NOT | CURRENT_DATE | CURRENT_TIMESTAMP - | VACUUM | RETAIN | HOURS ; SELECT: 'SELECT'; @@ -812,9 +810,6 @@ START: 'START'; TRANSACTION: 'TRANSACTION'; COMMIT: 'COMMIT'; ROLLBACK: 'ROLLBACK'; -VACUUM: 'VACUUM'; -RETAIN: 'RETAIN'; -HOURS: 'HOURS'; MACRO: 'MACRO'; IF: 'IF'; diff --git a/sql/core/src/main/antlr4/com/databricks/sql/acl/AclCommandBase.g4 b/sql/core/src/main/antlr4/com/databricks/sql/parser/DatabricksSqlBase.g4 similarity index 61% rename from sql/core/src/main/antlr4/com/databricks/sql/acl/AclCommandBase.g4 rename to sql/core/src/main/antlr4/com/databricks/sql/parser/DatabricksSqlBase.g4 index d372aeb41ddab..1472f30126bf4 100644 --- a/sql/core/src/main/antlr4/com/databricks/sql/acl/AclCommandBase.g4 +++ b/sql/core/src/main/antlr4/com/databricks/sql/parser/DatabricksSqlBase.g4 @@ -6,7 +6,31 @@ * License, Version 2.0, a copy of which you may obtain at * http://www.apache.org/licenses/LICENSE-2.0 */ -grammar AclCommandBase; +grammar DatabricksSqlBase; + +@members { + /** + * Verify whether current token is a valid decimal token (which contains dot). + * Returns true if the character that follows the token is not a digit or letter or underscore. + * + * For example: + * For char stream "2.3", "2." is not a valid decimal token, because it is followed by digit '3'. + * For char stream "2.3_", "2.3" is not a valid decimal token, because it is followed by '_'. + * For char stream "2.3W", "2.3" is not a valid decimal token, because it is followed by 'W'. + * For char stream "12.0D 34.E2+0.12 " 12.0D is a valid decimal token because it is folllowed + * by a space. 34.E2 is a valid decimal token because it is followed by symbol '+' + * which is not a digit or letter or underscore. + */ + public boolean isValidDecimal() { + int nextChar = _input.LA(1); + if (nextChar >= 'A' && nextChar <= 'Z' || nextChar >= '0' && nextChar <= '9' || + nextChar == '_') { + return false; + } else { + return true; + } + } +} tokens { DELIMITER @@ -17,11 +41,12 @@ singleStatement ; statement - : managePermissions #managePermissionsAlt - | ALTER securable OWNER TO identifier #alterOwner - | MSCK REPAIR securable PRIVILEGES #repairPrivileges - | SHOW GRANT identifier? ON (ALL| securable) #showPermissions - | .*? #passThrough + : managePermissions #managePermissionsAlt + | ALTER securable OWNER TO identifier #alterOwner + | MSCK REPAIR securable PRIVILEGES #repairPrivileges + | SHOW GRANT identifier? ON (ALL| securable) #showPermissions + | VACUUM (path=STRING | table=qualifiedName) (RETAIN number HOURS)? #vacuumTable + | .*? #passThrough ; managePermissions @@ -57,9 +82,15 @@ quotedIdentifier : BACKQUOTED_IDENTIFIER ; +number + : DECIMAL_VALUE #decimalLiteral + | INTEGER_VALUE #integerLiteral + ; + nonReserved - : ALTER | OWNER | TO | MSCK | REPAIR | PRIVILEGES | SHOW | GRANT | ON | ALL | WITH | OPTION | - REVOKE | FOR | FROM | CATALOG | DATABASE | TABLE | VIEW | FUNCTION | ANONYMOUS | FILE | ANY + : ALTER | OWNER | TO | MSCK | REPAIR | PRIVILEGES | SHOW | GRANT | ON | ALL | WITH | OPTION + | REVOKE | FOR | FROM | CATALOG | DATABASE | TABLE | VIEW | FUNCTION | ANONYMOUS | FILE | ANY + | VACUUM | RETAIN | HOURS ; ALTER: 'ALTER'; @@ -85,12 +116,24 @@ FUNCTION: 'FUNCTION'; ANONYMOUS: 'ANONYMOUS'; FILE: 'FILE'; ANY: 'ANY'; +VACUUM: 'VACUUM'; +RETAIN: 'RETAIN'; +HOURS: 'HOURS'; STRING : '\'' ( ~('\''|'\\') | ('\\' .) )* '\'' | '\"' ( ~('\"'|'\\') | ('\\' .) )* '\"' ; +INTEGER_VALUE + : DIGIT+ + ; + +DECIMAL_VALUE + : DIGIT+ EXPONENT + | DECIMAL_DIGITS EXPONENT? {isValidDecimal()}? + ; + IDENTIFIER : (LETTER | DIGIT | '_')+ ; @@ -99,6 +142,15 @@ BACKQUOTED_IDENTIFIER : '`' ( ~'`' | '``' )* '`' ; +fragment DECIMAL_DIGITS + : DIGIT+ '.' DIGIT* + | '.' DIGIT+ + ; + +fragment EXPONENT + : 'E' [+-]? DIGIT+ + ; + fragment DIGIT : [0-9] ; diff --git a/sql/core/src/main/scala/com/databricks/sql/acl/AclExtensions.scala b/sql/core/src/main/scala/com/databricks/sql/acl/AclExtensions.scala index d3e53c24988f7..dc00efaf661c6 100644 --- a/sql/core/src/main/scala/com/databricks/sql/acl/AclExtensions.scala +++ b/sql/core/src/main/scala/com/databricks/sql/acl/AclExtensions.scala @@ -11,6 +11,7 @@ package com.databricks.sql.acl import scala.util.control.NonFatal import com.databricks.sql.DatabricksStaticSQLConf +import com.databricks.sql.parser.DatabricksSqlParser import org.apache.spark.sql.{SparkSession, SparkSessionExtensions} import org.apache.spark.sql.catalog.BaseCatalogHooks @@ -57,7 +58,7 @@ class AclExtensions extends (SparkSessionExtensions => Unit) { } extensions.injectParser { (session, delegate) => if (isAclEnabled(session)) { - new AclCommandParser(client(session), delegate) + new DatabricksSqlParser(client(session), delegate) } else { delegate } diff --git a/sql/core/src/main/scala/com/databricks/sql/acl/CheckPermissions.scala b/sql/core/src/main/scala/com/databricks/sql/acl/CheckPermissions.scala index 4f3cd2d32bfa1..e163f52a43cc7 100644 --- a/sql/core/src/main/scala/com/databricks/sql/acl/CheckPermissions.scala +++ b/sql/core/src/main/scala/com/databricks/sql/acl/CheckPermissions.scala @@ -9,6 +9,7 @@ package com.databricks.sql.acl import com.databricks.sql.acl.Action._ +import com.databricks.sql.transaction.VacuumTableCommand import org.apache.spark.sql.catalog.{Catalog => PublicCatalog} import org.apache.spark.sql.catalyst.{FunctionIdentifier, TableIdentifier} diff --git a/sql/core/src/main/scala/com/databricks/sql/acl/AstCommandBuilder.scala b/sql/core/src/main/scala/com/databricks/sql/parser/DatabricksSqlCommandBuilder.scala similarity index 82% rename from sql/core/src/main/scala/com/databricks/sql/acl/AstCommandBuilder.scala rename to sql/core/src/main/scala/com/databricks/sql/parser/DatabricksSqlCommandBuilder.scala index 4822c2b43eb7d..96e7ed136e8cf 100644 --- a/sql/core/src/main/scala/com/databricks/sql/acl/AstCommandBuilder.scala +++ b/sql/core/src/main/scala/com/databricks/sql/parser/DatabricksSqlCommandBuilder.scala @@ -6,11 +6,13 @@ * License, Version 2.0, a copy of which you may obtain at * http://www.apache.org/licenses/LICENSE-2.0 */ -package com.databricks.sql.acl +package com.databricks.sql.parser import scala.collection.JavaConverters._ -import com.databricks.sql.acl.AclCommandBaseParser._ +import com.databricks.sql.acl._ +import com.databricks.sql.parser.DatabricksSqlBaseParser._ +import com.databricks.sql.transaction.VacuumTableCommand import org.apache.spark.sql.catalyst.{FunctionIdentifier, TableIdentifier} import org.apache.spark.sql.catalyst.parser.{ParseException, ParserUtils} @@ -19,8 +21,8 @@ import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan /** * Build an ACL related [[LogicalPlan]] from an ANTLR4 parser tree */ -class AstCommandBuilder(client: AclClient) - extends AclCommandBaseBaseVisitor[AnyRef] { +class DatabricksSqlCommandBuilder(client: AclClient) + extends DatabricksSqlBaseBaseVisitor[AnyRef] { import ParserUtils._ /** @@ -128,18 +130,18 @@ class AstCommandBuilder(client: AclClient) * Create a [[Securable]] object. */ override def visitSecurable(ctx: SecurableContext): Securable = withOrigin(ctx) { - Option(ctx.objectType).map(_.getType).getOrElse(AclCommandBaseParser.TABLE) match { - case AclCommandBaseParser.CATALOG => + Option(ctx.objectType).map(_.getType).getOrElse(DatabricksSqlBaseParser.TABLE) match { + case DatabricksSqlBaseParser.CATALOG => Catalog - case AclCommandBaseParser.DATABASE => + case DatabricksSqlBaseParser.DATABASE => Database(ctx.identifier.getText) - case AclCommandBaseParser.VIEW | AclCommandBaseParser.TABLE => + case DatabricksSqlBaseParser.VIEW | DatabricksSqlBaseParser.TABLE => Table(visitTableIdentifier(ctx.qualifiedName)) - case AclCommandBaseParser.FUNCTION if ctx.ANONYMOUS != null => + case DatabricksSqlBaseParser.FUNCTION if ctx.ANONYMOUS != null => AnonymousFunction - case AclCommandBaseParser.FUNCTION => + case DatabricksSqlBaseParser.FUNCTION => Function(visitFunctionIdentifier(ctx.qualifiedName)) - case AclCommandBaseParser.FILE => + case DatabricksSqlBaseParser.FILE => AnyFile case _ => throw new ParseException("Unknown Securable Object", ctx) @@ -170,6 +172,21 @@ class AstCommandBuilder(client: AclClient) } } + /** + * Create a [[VacuumTable]] logical plan. + * Example SQL : + * {{{ + * VACUUM ('/path/to/dir' | table_name) [RETAIN number HOURS]; + * }}} + */ + override def visitVacuumTable( + ctx: VacuumTableContext): LogicalPlan = withOrigin(ctx) { + VacuumTableCommand( + Option(ctx.path).map(string), + Option(ctx.table).map(visitTableIdentifier), + Option(ctx.number).map(_.getText.toDouble)) + } + /** * Return null for every other query. These queries should be passed to a delegate parser. */ diff --git a/sql/core/src/main/scala/com/databricks/sql/acl/AclCommandParser.scala b/sql/core/src/main/scala/com/databricks/sql/parser/DatabricksSqlParser.scala similarity index 85% rename from sql/core/src/main/scala/com/databricks/sql/acl/AclCommandParser.scala rename to sql/core/src/main/scala/com/databricks/sql/parser/DatabricksSqlParser.scala index 284772de682b1..72eb420ba49cb 100644 --- a/sql/core/src/main/scala/com/databricks/sql/acl/AclCommandParser.scala +++ b/sql/core/src/main/scala/com/databricks/sql/parser/DatabricksSqlParser.scala @@ -6,14 +6,15 @@ * License, Version 2.0, a copy of which you may obtain at * http://www.apache.org/licenses/LICENSE-2.0 */ -package com.databricks.sql.acl +package com.databricks.sql.parser -import com.databricks.sql.acl.AclCommandBaseParser._ +import com.databricks.sql.acl.AclClient +import com.databricks.sql.parser.DatabricksSqlBaseParser._ import org.antlr.v4.runtime._ import org.antlr.v4.runtime.atn.PredictionMode import org.antlr.v4.runtime.misc.ParseCancellationException -import org.apache.spark.sql.AnalysisException +import org.apache.spark.sql.{AnalysisException, Dataset, SparkSession} import org.apache.spark.sql.catalyst.TableIdentifier import org.apache.spark.sql.catalyst.expressions.Expression import org.apache.spark.sql.catalyst.parser._ @@ -25,9 +26,9 @@ import org.apache.spark.sql.types.DataType * Parser for ACL related commands. The parser passes the query to an underlying (more complete) * parser if it cannot parse the query. */ -class AclCommandParser(client: AclClient, delegate: ParserInterface) extends ParserInterface { +class DatabricksSqlParser(client: AclClient, delegate: ParserInterface) extends ParserInterface { - val builder = new AstCommandBuilder(client) + val builder = new DatabricksSqlCommandBuilder(client) override def parseDataType(sqlText: String): DataType = delegate.parseDataType(sqlText) @@ -46,13 +47,13 @@ class AclCommandParser(client: AclClient, delegate: ParserInterface) extends Par } } - protected def parse[T](command: String)(toResult: AclCommandBaseParser => T): T = { - val lexer = new AclCommandBaseLexer(new ANTLRNoCaseStringStream(command)) + protected def parse[T](command: String)(toResult: DatabricksSqlBaseParser => T): T = { + val lexer = new DatabricksSqlBaseLexer(new ANTLRNoCaseStringStream(command)) lexer.removeErrorListeners() lexer.addErrorListener(ParseErrorListener) val tokenStream = new CommonTokenStream(lexer) - val parser = new AclCommandBaseParser(tokenStream) + val parser = new DatabricksSqlBaseParser(tokenStream) parser.addParseListener(PostProcessor) parser.removeErrorListeners() parser.addErrorListener(ParseErrorListener) @@ -96,7 +97,7 @@ class ANTLRNoCaseStringStream(input: String) extends ANTLRInputStream(input) { /** * The post-processor validates & cleans-up the parse tree during the parse process. */ -case object PostProcessor extends AclCommandBaseBaseListener { +case object PostProcessor extends DatabricksSqlBaseBaseListener { /** Remove the back ticks from an Identifier. */ override def exitQuotedIdentifier(ctx: QuotedIdentifierContext): Unit = { @@ -121,7 +122,7 @@ case object PostProcessor extends AclCommandBaseBaseListener { val token = ctx.getChild(0).getPayload.asInstanceOf[Token] parent.addChild(f(new CommonToken( new org.antlr.v4.runtime.misc.Pair(token.getTokenSource, token.getInputStream), - AclCommandBaseParser.IDENTIFIER, + DatabricksSqlBaseParser.IDENTIFIER, token.getChannel, token.getStartIndex + stripMargins, token.getStopIndex - stripMargins))) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/VacuumTableCommand.scala b/sql/core/src/main/scala/com/databricks/sql/transaction/VacuumTableCommand.scala similarity index 96% rename from sql/core/src/main/scala/org/apache/spark/sql/execution/command/VacuumTableCommand.scala rename to sql/core/src/main/scala/com/databricks/sql/transaction/VacuumTableCommand.scala index 10b3050f58590..cbeb45f3025ef 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/VacuumTableCommand.scala +++ b/sql/core/src/main/scala/com/databricks/sql/transaction/VacuumTableCommand.scala @@ -15,7 +15,7 @@ * limitations under the License. */ -package org.apache.spark.sql.execution.command +package com.databricks.sql.transaction import java.net.URI @@ -24,6 +24,7 @@ import org.apache.hadoop.fs.Path import org.apache.spark.sql.{Row, SparkSession} import org.apache.spark.sql.catalyst.TableIdentifier import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference} +import org.apache.spark.sql.execution.command.RunnableCommand import org.apache.spark.sql.transaction.DatabricksAtomicCommitProtocol import org.apache.spark.sql.types._ diff --git a/sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala b/sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala index fe1f3ddba224b..1b5a1d9143ca5 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala @@ -80,6 +80,10 @@ class SparkSession private( this(sc, None, new SparkSessionExtensions) } + private[sql] def this(sc: SparkContext, extensions: Option[SparkSessionExtensions]) { + this(sc, None, extensions.getOrElse(new SparkSessionExtensions)) + } + sparkContext.assertNotStopped() /** diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala index ced6b68d33fa8..67d9a4fb33cce 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala @@ -86,21 +86,6 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder { ResetCommand } - /** - * Create a [[VacuumTable]] logical plan. - * Example SQL : - * {{{ - * VACUUM ('/path/to/dir' | table_name) [RETAIN number HOURS]; - * }}} - */ - override def visitVacuumTable( - ctx: VacuumTableContext): LogicalPlan = withOrigin(ctx) { - VacuumTableCommand( - Option(ctx.path).map(string), - Option(ctx.tableIdentifier).map(visitTableIdentifier), - Option(ctx.number).map(_.getText.toDouble)) - } - /** * Create an [[AnalyzeTableCommand]] command or an [[AnalyzeColumnCommand]] command. * Example SQL for analyzing table : diff --git a/sql/core/src/test/scala/com/databricks/sql/acl/CheckPermissionRuleSuite.scala b/sql/core/src/test/scala/com/databricks/sql/acl/CheckPermissionRuleSuite.scala index a6100ea6e950b..d9cf97b9e48b8 100644 --- a/sql/core/src/test/scala/com/databricks/sql/acl/CheckPermissionRuleSuite.scala +++ b/sql/core/src/test/scala/com/databricks/sql/acl/CheckPermissionRuleSuite.scala @@ -9,6 +9,7 @@ package com.databricks.sql.acl import com.databricks.sql.acl.Action.{ReadMetadata, Select} +import com.databricks.sql.transaction.VacuumTableCommand import org.apache.hadoop.fs.Path import org.mockito.Mockito._ diff --git a/sql/core/src/test/scala/com/databricks/sql/acl/AclCommandParseSuite.scala b/sql/core/src/test/scala/com/databricks/sql/parser/DatabricksSqlCommandParserSuite.scala similarity index 88% rename from sql/core/src/test/scala/com/databricks/sql/acl/AclCommandParseSuite.scala rename to sql/core/src/test/scala/com/databricks/sql/parser/DatabricksSqlCommandParserSuite.scala index 63da9655024e6..57eeab9711e64 100644 --- a/sql/core/src/test/scala/com/databricks/sql/acl/AclCommandParseSuite.scala +++ b/sql/core/src/test/scala/com/databricks/sql/parser/DatabricksSqlCommandParserSuite.scala @@ -6,9 +6,11 @@ * License, Version 2.0, a copy of which you may obtain at * http://www.apache.org/licenses/LICENSE-2.0 */ -package com.databricks.sql.acl +package com.databricks.sql.parser +import com.databricks.sql.acl._ import com.databricks.sql.acl.Action._ +import com.databricks.sql.transaction.VacuumTableCommand import org.apache.spark.sql.catalyst.{FunctionIdentifier, SimpleCatalystConf, TableIdentifier} import org.apache.spark.sql.catalyst.analysis.FunctionRegistry @@ -19,7 +21,7 @@ import org.apache.spark.sql.catalyst.parser.{CatalystSqlParser, ParseException} import org.apache.spark.sql.catalyst.plans.PlanTest import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan -class AclCommandParseSuite extends PlanTest { +class DatabricksSqlCommandParserSuite extends PlanTest { val client = NoOpAclClient val catalog = new SessionCatalog( @@ -27,7 +29,7 @@ class AclCommandParseSuite extends PlanTest { FunctionRegistry.builtin, SimpleCatalystConf(caseSensitiveAnalysis = false)) - val parser = new AclCommandParser(client, CatalystSqlParser) + val parser = new DatabricksSqlParser(client, CatalystSqlParser) def checkAnswer(query: String, plan: LogicalPlan): Unit = { comparePlans(parser.parsePlan(query), plan) @@ -44,6 +46,7 @@ class AclCommandParseSuite extends PlanTest { val testTbl1 = Table(TableIdentifier("tbl1", Option("test"))) val tbl1 = Table(TableIdentifier("tbl1")) val tblCatalog = Table(TableIdentifier("catalog")) + val testPath = "/test/path" val functionReflect = Function(FunctionIdentifier("reflect")) @@ -180,6 +183,30 @@ class AclCommandParseSuite extends PlanTest { intercept("MSCK REPAIR KEYBOARD bubba PRIVILEGES") } + test("vacuum") { + checkAnswer( + s"VACUUM '$testPath'", + VacuumTableCommand(Some(testPath), None, None)) + + checkAnswer( + s"""VACUUM "$testPath" RETAIN 1.25 HOURS""", + VacuumTableCommand(Some(testPath), None, Some(1.25))) + + checkAnswer( + "VACUUM `tbl1`", + VacuumTableCommand(None, Some(tbl1.key), None)) + + checkAnswer( + "VACUUM test.tbl1 RETAIN 2 HOURS", + VacuumTableCommand(None, Some(testTbl1.key), Some(2.0))) + + intercept("VACUUM") + + intercept("VACUUM TABLE `tbl1`") + + intercept(s"VACUUM $testPath") // because the given path is unquoted + } + test("pass through") { checkAnswer("select * from a union all select * from b", table("a").select(star()).union(table("b").select(star()))) diff --git a/sql/core/src/test/scala/com/databricks/sql/transaction/DatabricksAtomicCommitProtocolSuite.scala b/sql/core/src/test/scala/com/databricks/sql/transaction/DatabricksAtomicCommitProtocolSuite.scala index d49d6a2d627d3..b4e938666d1aa 100644 --- a/sql/core/src/test/scala/com/databricks/sql/transaction/DatabricksAtomicCommitProtocolSuite.scala +++ b/sql/core/src/test/scala/com/databricks/sql/transaction/DatabricksAtomicCommitProtocolSuite.scala @@ -13,16 +13,27 @@ import java.io._ import scala.collection.mutable import com.databricks.sql.DatabricksSQLConf._ +import com.databricks.sql.acl.NoOpAclClient +import com.databricks.sql.parser.DatabricksSqlParser import org.apache.hadoop.conf.Configuration import org.apache.hadoop.fs._ -import org.apache.spark.SparkEnv -import org.apache.spark.sql.QueryTest +import org.apache.spark.{DebugFilesystem, SparkEnv} +import org.apache.spark.sql.{QueryTest, SparkSessionExtensions} import org.apache.spark.sql.execution.datasources.InMemoryFileIndex -import org.apache.spark.sql.test.SharedSQLContext +import org.apache.spark.sql.test.{SharedSQLContext, TestSparkSession} import org.apache.spark.util.{Clock, ManualClock, SystemClock} class DatabricksAtomicCommitProtocolSuite extends QueryTest with SharedSQLContext { + override def createSparkSession: TestSparkSession = { + val extensions = new SparkSessionExtensions + extensions.injectParser((_, delegate) => new DatabricksSqlParser(NoOpAclClient, delegate)) + + new TestSparkSession( + sparkConf.set("spark.hadoop.fs.file.impl", classOf[DebugFilesystem].getName), + Some(extensions)) + } + test("read protocol ignores uncommitted jobs") { withTempDir { dir => create(dir, "_started_12345") diff --git a/sql/core/src/test/scala/org/apache/spark/sql/test/TestSQLContext.scala b/sql/core/src/test/scala/org/apache/spark/sql/test/TestSQLContext.scala index 1bdf956247f69..223c45c29eeb8 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/test/TestSQLContext.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/test/TestSQLContext.scala @@ -18,16 +18,26 @@ package org.apache.spark.sql.test import org.apache.spark.{SparkConf, SparkContext} -import org.apache.spark.sql.SparkSession +import org.apache.spark.sql.{SparkSession, SparkSessionExtensions} import org.apache.spark.sql.internal.{SessionState, SQLConf} /** * A special [[SparkSession]] prepared for testing. */ -private[sql] class TestSparkSession(sc: SparkContext) extends SparkSession(sc) { self => - def this(sparkConf: SparkConf) { +private[sql] class TestSparkSession(sc: SparkContext, extensions: Option[SparkSessionExtensions]) + extends SparkSession(sc, extensions) { self => + + private[sql] def this(sc: SparkContext) { + this(sc, None) + } + + def this(sparkConf: SparkConf, extensions: Option[SparkSessionExtensions]) { this(new SparkContext("local[2]", "test-sql-context", - sparkConf.set("spark.sql.testkey", "true"))) + sparkConf.set("spark.sql.testkey", "true")), extensions) + } + + def this(sparkConf: SparkConf) { + this(sparkConf, None) } def this() {