[SC-5603] Make CTAS work with ACL enable DB Spark

## What changes were proposed in this pull request?
ACL enabled Spark currently throws a table not found exception when we issue a CTAS command. This bug is caused by a number of factors:

1. CTAS is composed of two stages: creating the table and inserting data into the table.
2. The order of these commands has been inverted to make sure we never expose a partially written table to an end user.
3. We check ACLs for the CTAS command, and also for its child commands. ACL checking fails because we cannot find the table when executing the insert (it has not been created yet).

This PR fixes this by only ACL checking the CTAS command, and not its children. This fix is also applied to all other `RunnableCommand`s. The only exception is the `ExplainCommand`, in this case we **want** to execute `CheckAnalysis` on its child.

This has been implemented by adding a trusted mode to `CheckPermissions`. We do not check the permissions when we are in trusted mode. The value of trusted mode is thread local, and can by set by calling the `CheckPermissions.trusted` method. Note that a creative developer can abuse this by either using Java (the trusted method is package-local to `com.databricks`) or some light reflection. The other part of the solution is that we add a planning rule to `QueryExecution`, which wraps a `RunnableCommand` with a `TrustedRunnableCommand `. Note that the downside to this is that we are introducing a change in an apache package which might create merge conflicts.

Finally I have added an sbt task `sparkShellACL` which starts the Spark Shell with ACLs enabled. This is easier for testing. Remember to set `sc.setLocalProperty("spark.databricks.token", "super")` when you start the shell (if you don't every command will fail). You can change users by modifying the `spark.databricks.token` local spark properties.

## How was this patch tested?
Existing tests & Added an integration test to the ACL end to end tests.

Author: Herman van Hovell <hvanhovell@databricks.com>

Closes apache#184 from hvanhovell/SC-5603.

Author: hvanhovell

project/SparkBuild.scala

@@ -416,6 +416,8 @@ object SparkBuild extends PomBuild {
      """.stripMargin)
   val sparkSql = taskKey[Unit]("starts the spark sql CLI.")
 
+  val sparkShellAcl = taskKey[Unit]("start a spark-shell with ACL support.")
+
   enable(Seq(
     connectInput in run := true,
     fork := true,
@@ -427,6 +429,15 @@ object SparkBuild extends PomBuild {
       (runMain in Compile).toTask(" org.apache.spark.repl.Main -usejavacp").value
     },
 
+    sparkShellAcl := {
+      (runMain in Compile).toTask(
+        " -Dspark.sql.extensions=com.databricks.sql.acl.HiveAclExtensions" +
+        " -Dspark.databricks.acl.provider=com.databricks.sql.acl.ReflectionBackedAclProvider" +
+        " -Dspark.databricks.acl.client=com.databricks.sql.acl.SharedAclBackend" +
+        " -Dspark.databricks.acl.enabled=true" +
+        " org.apache.spark.repl.Main -usejavacp").value
+    },
+
     sparkPackage := {
       import complete.DefaultParsers._
       val packages :: className :: otherArgs = spaceDelimited("<group:artifact:version> <MainClass> [args]").parsed.toList

sql/core/src/main/scala/com/databricks/sql/acl/CheckPermissions.scala

@@ -32,6 +32,9 @@ class CheckPermissions(catalog: PublicCatalog, aclClient: AclClient)
   val permissionChecker = new PermissionChecker(aclClient)
 
   def apply(plan: LogicalPlan): Unit = {
+    if (CheckPermissions.isTrusted) {
+      return
+    }
     getRequestsToCheck(plan).foreach { request =>
       if (!permissionChecker(request)) {
         throw new SecurityException(toErrorMessageForFailedRequest(request))
@@ -373,3 +376,18 @@ class CheckPermissions(catalog: PublicCatalog, aclClient: AclClient)
     s"User does not ${actionString} ${request.securable.prettyString}"
   }
 }
+
+private[databricks] object CheckPermissions {
+  private val trusted = new ThreadLocal[Boolean] {
+    override def initialValue(): Boolean = false
+  }
+
+  private[databricks] def isTrusted: Boolean = trusted.get()
+
+  private[databricks] def trusted[T](block: => T): T = {
+    trusted.set(true)
+    try block finally {
+      trusted.set(false)
+    }
+  }
+}

sql/core/src/test/scala/com/databricks/sql/acl/SharedAclBackend.scala renamed to sql/core/src/main/scala/com/databricks/sql/acl/SharedAclBackend.scala

@@ -138,6 +138,11 @@ class SharedAclBackend {
     }
     log.info(s"modify[token=$token, modifications=${modifications.mkString("[", ",", "]")}]")
   }
+
+  /**
+   * Get a token from the Databricks DriverLocal. This is not implemented.
+   */
+  def getTokenFromLocalContext(): Option[String] = None
 }
 
 object SharedAclBackend {

sql/core/src/main/scala/com/databricks/sql/acl/TrustedRunnableCommand.scala

@@ -0,0 +1,54 @@
+/*
+ * Copyright (C) 2016 Databricks, Inc.
+ *
+ * Portions of this software incorporate or are derived from software contained within Apache Spark,
+ * and this modified software differs from the Apache Spark software provided under the Apache
+ * 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
+
+import org.apache.spark.sql.{Row, SparkSession}
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.execution.SparkPlan
+import org.apache.spark.sql.execution.command._
+
+/**
+ * A trusted runnable command that we trust as soon its permission check out. In practice this
+ * means that we do not check permissions for the command or for commands that the command issues
+ * when the command is executed. An example of the latter is a CreateTableAsSelect command that
+ * both issues a create and an insert command.
+ */
+case class TrustedRunnableCommand(cmd: RunnableCommand) extends RunnableCommand {
+  assert(cmd != null)
+
+  override def output: Seq[Attribute] = cmd.output
+
+  override def run(sparkSession: SparkSession): Seq[Row] = {
+    CheckPermissions.trusted(cmd.run(sparkSession))
+  }
+
+  override def generateTreeString(
+      depth: Int,
+      lastChildren: Seq[Boolean],
+      builder: StringBuilder,
+      verbose: Boolean,
+      prefix: String): StringBuilder = {
+    cmd.generateTreeString(depth, lastChildren, builder, verbose, prefix)
+  }
+}
+
+/**
+ * This rule wraps a RunnableCommand with a TrustedRunnableCommand before the command is executed.
+ */
+object MakeRunnableCommandTrusted extends Rule[SparkPlan] {
+  override def apply(plan: SparkPlan): SparkPlan = plan match {
+    case ExecutedCommandExec(_: TrustedRunnableCommand) => plan
+    case ExecutedCommandExec(_: ExplainCommand) => plan
+    case ExecutedCommandExec(_: ShowTablesCommand) => plan
+    case ExecutedCommandExec(_: DescribeTableCommand) => plan
+    case ExecutedCommandExec(cmd) => ExecutedCommandExec(TrustedRunnableCommand(cmd))
+    case _ => plan
+  }
+}

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala

@@ -245,7 +245,12 @@ class Dataset[T] private[sql](
    */
   private[sql] def showString(_numRows: Int, truncate: Int = 20): String = {
     val numRows = _numRows.max(0)
-    val takeResult = toDF().take(numRows + 1)
+    val takeResult = withAction("show", limit(numRows + 1).queryExecution) { plan =>
+      val encoder = RowEncoder(schema).resolveAndBind(
+        logicalPlan.output,
+        sparkSession.sessionState.analyzer)
+      plan.executeCollect().map(encoder.fromRow)
+    }
     val hasMoreData = takeResult.length > numRows
     val data = takeResult.take(numRows)
 

sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala

@@ -20,6 +20,8 @@ package org.apache.spark.sql.execution
 import java.nio.charset.StandardCharsets
 import java.sql.Timestamp
 
+import com.databricks.sql.acl.MakeRunnableCommandTrusted
+
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.{AnalysisException, Row, SparkSession}
 import org.apache.spark.sql.catalyst.InternalRow
@@ -101,7 +103,8 @@ class QueryExecution(val sparkSession: SparkSession, val logical: LogicalPlan) {
     EnsureRequirements(sparkSession.sessionState.conf),
     CollapseCodegenStages(sparkSession.sessionState.conf),
     ReuseExchange(sparkSession.sessionState.conf),
-    ReuseSubquery(sparkSession.sessionState.conf))
+    ReuseSubquery(sparkSession.sessionState.conf),
+    MakeRunnableCommandTrusted)
 
   protected def stringOrError[A](f: => A): String =
     try f.toString catch { case e: AnalysisException => e.toString }

sql/hive-thriftserver/src/test/resources/acl-tests/inputs/command.sql

@@ -0,0 +1,7 @@
+create database db1;
+
+create table db1.test using parquet as select * from range(100);
+
+drop table db1.test;
+
+drop database db1;

sql/hive-thriftserver/src/test/resources/acl-tests/results/command.sql.out

@@ -0,0 +1,42 @@
+-- Automatically generated by ThriftEndToEndAclTestSuite
+-- Number of queries: 4
+
+
+-- !query 0
+create database db1
+-- !query 0 token
+super
+-- !query 0 schema
+struct<Result:string>
+-- !query 0 output
+
+
+
+-- !query 1
+create table db1.test using parquet as select * from range(100)
+-- !query 1 token
+super
+-- !query 1 schema
+struct<Result:string>
+-- !query 1 output
+
+
+
+-- !query 2
+drop table db1.test
+-- !query 2 token
+super
+-- !query 2 schema
+struct<Result:string>
+-- !query 2 output
+
+
+
+-- !query 3
+drop database db1
+-- !query 3 token
+super
+-- !query 3 schema
+struct<Result:string>
+-- !query 3 output
+