[CARMEL-3172] Upload Support (#62)

* [CARMEL-3172] Upload Support

* Fix code style

* fix ut

* fix ut

Author: xingchaozh

sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4

@@ -227,6 +227,8 @@ statement
     | CLEAR CACHE                                                      #clearCache
     | LOAD DATA LOCAL? INPATH path=STRING OVERWRITE? INTO TABLE
         multipartIdentifier partitionSpec?                             #loadData
+    | UPLOAD DATA INPATH path=STRING OVERWRITE? INTO TABLE
+        tableIdentifier partitionSpec? optionSpec?                     #uploadData
     | TRUNCATE TABLE multipartIdentifier partitionSpec?                #truncateTable
     | MSCK REPAIR TABLE multipartIdentifier                            #repairTable
     | op=(ADD | LIST) identifier (STRING | .*?)                        #manageResource
@@ -254,6 +256,13 @@ statement
     | unsupportedHiveNativeCommands .*?                                #failNativeCommand
     ;
 
+optionSpec
+    : OPTION '(' optionVal (',' optionVal)* ')'
+    ;
+
+optionVal
+    : identifier (EQ constant)?
+    ;
 
 privilegeList
     : privilegeDef (',' privilegeDef)*
@@ -1248,6 +1257,7 @@ ansiNonReserved
     | UNLOCK
     | UNSET
     | UPDATE
+    | UPLOAD
     | USE
     | VALUES
     | VIEW
@@ -1789,6 +1799,7 @@ UNKNOWN: 'UNKNOWN';
 UNLOCK: 'UNLOCK';
 UNSET: 'UNSET';
 UPDATE: 'UPDATE';
+UPLOAD: 'UPLOAD';
 USE: 'USE';
 USER: 'USER';
 USING: 'USING';

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala

@@ -529,6 +529,31 @@ class AstBuilder(conf: SQLConf) extends SqlBaseBaseVisitor[AnyRef] with Logging
     }
   }
 
+  /**
+   * Create a option specification map.
+   */
+  override def visitOptionSpec(
+        ctx: OptionSpecContext): Map[String, Option[String]] = withOrigin(ctx) {
+    val opts = ctx.optionVal.asScala.map { pVal =>
+      val name = pVal.identifier.getText
+      val value = Option(pVal.constant).map(visitStringConstant)
+      name -> value
+    }
+    checkDuplicateKeys(opts, ctx)
+    opts.toMap
+  }
+
+  /**
+   * Create a option specification map without optional values.
+   */
+  protected def visitNonOptionalOptionSpec(
+        ctx: OptionSpecContext): Map[String, String] = withOrigin(ctx) {
+    visitOptionSpec(ctx).map {
+      case (key, None) => throw new ParseException(s"Found an empty option key '$key'.", ctx)
+      case (key, Some(value)) => key -> value
+    }
+  }
+
   /**
    * Convert a constant of any type into a string. This is typically used in DDL commands, and its
    * main purpose is to prevent slight differences due to back to back conversions i.e.:
@@ -3310,6 +3335,19 @@ class AstBuilder(conf: SQLConf) extends SqlBaseBaseVisitor[AnyRef] with Logging
     )
   }
 
+  override def visitUploadData(ctx: UploadDataContext): LogicalPlan = withOrigin(ctx) {
+    if (!conf.hiveThriftServerDataUploadEnabled) {
+      throw new ParseException(s"UPLOAD DATA is not supported", ctx)
+    }
+    UploadDataStatement(
+      table = visitTableIdentifier(ctx.tableIdentifier),
+      path = string(ctx.path),
+      isOverwrite = ctx.OVERWRITE != null,
+      partitionSpec = Option(ctx.partitionSpec).map(visitNonOptionalPartitionSpec),
+      optionSpec = Option(ctx.optionSpec()).map(visitNonOptionalOptionSpec)
+    )
+  }
+
   /**
    * Creates a [[ShowCreateTableStatement]]
    */

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statements.scala

@@ -390,6 +390,16 @@ case class LoadDataStatement(
     isOverwrite: Boolean,
     partition: Option[TablePartitionSpec]) extends ParsedStatement
 
+/**
+ * A UPLOAD DATA INTO TABLE statement, as parsed from SQL
+ */
+case class UploadDataStatement(
+    table: TableIdentifier,
+    path: String,
+    isOverwrite: Boolean,
+    partitionSpec: Option[TablePartitionSpec],
+    optionSpec: Option[Map[String, String]]) extends ParsedStatement
+
 /**
  * A SHOW CREATE TABLE statement, as parsed from SQL.
  */

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -3348,6 +3348,24 @@ class SQLConf extends Serializable with Logging {
   def hiveThriftServerEventLogDir: String =
     getConf(StaticSQLConf.HIVE_THRIFT_SERVER_EVENTLOG_DIR)
 
+  def hiveThriftServerDataUploadEnabled: Boolean =
+    getConf(StaticSQLConf.HIVE_THRIFT_SERVER_DATA_UPLOAD_ENABLED)
+
+  def hiveThriftServerDataUploadDynamicPartitionEnabled: Boolean =
+    getConf(StaticSQLConf.HIVE_THRIFT_SERVER_DATA_UPLOAD_DYNAMIC_PARTITION_ENABLED)
+
+  def hiveThriftServerDataUploadTemporaryBaseDir: String =
+    getConf(StaticSQLConf.HIVE_THRIFT_SERVER_DATA_UPLOAD_TEMPORARY_BASE_DIR)
+
+  def hiveThriftServerDataUploadTemporaryFileMaxSize: Long =
+    getConf(StaticSQLConf.HIVE_THRIFT_SERVER_DATA_UPLOAD_TEMPORARY_FILE_MAX_SIZE)
+
+  def hiveThriftServerDataUploadWorkspaceBaseDir: String =
+    getConf(StaticSQLConf.HIVE_THRIFT_SERVER_DATA_UPLOAD_WORKSPACE_BASE_DIR)
+
+  def hiveThriftServerDataUploadNotAllowedCsvOptions: String =
+    getConf(StaticSQLConf.HIVE_THRIFT_SERVER_DATA_UPLOAD_NOT_ALLOWED_CSV_OPTIONS)
+
   def orderByOrdinal: Boolean = getConf(ORDER_BY_ORDINAL)
 
   def groupByOrdinal: Boolean = getConf(GROUP_BY_ORDINAL)

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/StaticSQLConf.scala

@@ -132,13 +132,48 @@ object StaticSQLConf {
       .doc("Path to record event log.")
       .stringConf
       .createWithDefault("/tmp/spark-events")
-
   val HIVE_THRIFT_SERVER_DATA_DOWNLOAD_MAX_SIZE =
     buildStaticConf("spark.sql.hive.thriftServer.data.download.max.size")
       .doc("The maximum data size allowed downloaded from Thrift server.")
       .bytesConf(ByteUnit.BYTE)
       .createWithDefault(100L * 1024 * 1024 * 1024) // 100G
 
+  val HIVE_THRIFT_SERVER_DATA_UPLOAD_ENABLED =
+    buildStaticConf("spark.sql.hive.thriftServer.data.upload.enabled")
+      .doc("Enable data upload")
+      .booleanConf
+      .createWithDefault(true)
+
+  val HIVE_THRIFT_SERVER_DATA_UPLOAD_DYNAMIC_PARTITION_ENABLED =
+    buildStaticConf("spark.sql.hive.thriftServer.data.upload.dynamic.partition.enabled")
+      .doc("Enable data upload for dynamic partition insert")
+      .booleanConf
+      .createWithDefault(true)
+
+  val HIVE_THRIFT_SERVER_DATA_UPLOAD_TEMPORARY_BASE_DIR =
+    buildStaticConf("spark.sql.hive.thriftServer.data.upload.temporary.base.dir")
+      .doc("Temporary path for data uploading.")
+      .stringConf
+      .createWithDefault("/tmp/spark-uploads")
+
+  val HIVE_THRIFT_SERVER_DATA_UPLOAD_TEMPORARY_FILE_MAX_SIZE =
+    buildStaticConf("spark.sql.hive.thriftServer.data.upload.temporary.file.max.size")
+      .doc("The max size of file uploaded in bytes for data uploading.")
+      .longConf
+      .createWithDefault(200*1024*1024)
+
+  val HIVE_THRIFT_SERVER_DATA_UPLOAD_WORKSPACE_BASE_DIR =
+    buildStaticConf("spark.sql.hive.thriftServer.data.upload.workspace.base.dir")
+      .doc("The base directory of workspaces for data uploading.")
+      .stringConf
+      .createWithDefault("/tmp/spark-workspaces")
+
+  val HIVE_THRIFT_SERVER_DATA_UPLOAD_NOT_ALLOWED_CSV_OPTIONS =
+    buildStaticConf("spark.sql.hive.thriftServer.data.upload.not.allowed.csv.options")
+      .doc("The not allowed csv options for data uploading.")
+      .stringConf
+      .createWithDefault("path")
+
   val SPARK_SESSION_EXTENSIONS = buildStaticConf("spark.sql.extensions")
     .doc("A comma-separated list of classes that implement " +
       "Function1[SparkSessionExtensions, Unit] used to configure Spark Session extensions. The " +

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala

@@ -20,6 +20,7 @@ package org.apache.spark.sql.catalyst.parser
 import java.util.Locale
 
 import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.analysis.{AnalysisTest, GlobalTempView, LocalTempView, PersistedView, UnresolvedAttribute, UnresolvedNamespace, UnresolvedRelation, UnresolvedStar, UnresolvedTable, UnresolvedTableOrView}
 import org.apache.spark.sql.catalyst.catalog.{ArchiveResource, BucketSpec, FileResource, FunctionResource, FunctionResourceType, JarResource}
 import org.apache.spark.sql.catalyst.expressions.{EqualTo, Literal, PartitioningAttribute}
@@ -1557,6 +1558,33 @@ class DDLParserSuite extends AnalysisTest {
         Some(Map("ds" -> "2017-06-10"))))
   }
 
+  test("UPLOAD DATA INTO table") {
+    comparePlans(
+      parsePlan(
+        s"""
+           |UPLOAD DATA INPATH 'path' INTO TABLE table1
+         """.stripMargin),
+      UploadDataStatement(
+        new TableIdentifier("table1", None),
+        "path",
+        false,
+        None,
+        None))
+
+    comparePlans(
+      parsePlan(
+        s"""
+           |UPLOAD DATA INPATH 'path' OVERWRITE INTO
+           |TABLE table1 PARTITION(c='1', d='2') OPTION(e='3', f='true')
+         """.stripMargin),
+      UploadDataStatement(
+        new TableIdentifier("table1", None),
+        "path",
+        true,
+        Some(Map("c" -> "1", "d" -> "2")),
+        Some(Map("e" -> "3", "f" -> "true"))))
+  }
+
   test("SHOW CREATE table") {
     comparePlans(
       parsePlan("SHOW CREATE TABLE a.b.c"),

sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala

@@ -402,6 +402,14 @@ class ResolveSessionCatalog(
         isOverwrite,
         partition)
 
+    case UploadDataStatement(tbl, path, isOverwrite, partitionSpec, optionSpec) =>
+      UploadDataCommand(
+        tbl,
+        path,
+        isOverwrite,
+        partitionSpec,
+        optionSpec)
+
     case ShowCreateTableStatement(tbl, asSerde) if !asSerde =>
       val name = parseTempViewOrV1Table(tbl, "SHOW CREATE TABLE")
       ShowCreateTableCommand(name.asTableIdentifier)