bulkload parser

sql/hbase/src/main/scala/org/apache/spark/sql/hbase/HBaseSQLParser.scala

@@ -20,9 +20,15 @@ import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.catalyst.{SqlLexical, SqlParser}
 import org.apache.spark.sql.catalyst.SparkSQLParser
-import org.apache.spark.sql.hbase.logical.{CreateHBaseTablePlan, DropTablePlan}
+import org.apache.spark.sql.hbase.logical.{CreateHBaseTablePlan, DropTablePlan, LoadDataIntoTablePlan}
 
 class HBaseSQLParser extends SqlParser {
+
+  protected val DATA = Keyword("DATA")
+  protected val LOAD = Keyword("LOAD")
+  protected val LOCAL = Keyword("LOCAL")
+  protected val INPATH = Keyword("INPATH")
+
   protected val BULK = Keyword("BULK")
   protected val CREATE = Keyword("CREATE")
   protected val DROP = Keyword("DROP")
@@ -55,7 +61,7 @@ class HBaseSQLParser extends SqlParser {
         | EXCEPT ^^^ { (q1: LogicalPlan, q2: LogicalPlan) => Except(q1, q2)}
         | UNION ~ DISTINCT.? ^^^ { (q1: LogicalPlan, q2: LogicalPlan) => Distinct(Union(q1, q2))}
         )
-      | insert | create | drop | alter
+      | insert | create | drop | alter | load
       )
 
   override protected lazy val insert: Parser[LogicalPlan] =
@@ -137,6 +143,18 @@ class HBaseSQLParser extends SqlParser {
       case tn ~ op ~ tc ~ cf => null
     }
 
+  protected lazy val load: Parser[LogicalPlan] =
+  (
+    (LOAD ~> DATA ~> INPATH ~> stringLit) ~
+    (opt(OVERWRITE) ~> INTO ~> TABLE ~> relation) ^^ {
+      case filePath ~ table => LoadDataIntoTablePlan(filePath, table, false)
+    }
+  | (LOAD ~> DATA ~> LOCAL ~> INPATH ~> stringLit) ~
+      (opt(OVERWRITE) ~> INTO ~> TABLE ~> relation) ^^ {
+      case filePath ~ table => LoadDataIntoTablePlan(filePath, table, true)
+    }
+  )
+
   protected lazy val tableCol: Parser[(String, String)] =
     ident ~ (STRING | BYTE | SHORT | INTEGER | LONG | FLOAT | DOUBLE | BOOLEAN) ^^ {
       case e1 ~ e2 => (e1, e2)

sql/hbase/src/main/scala/org/apache/spark/sql/hbase/logical/HBaseLogicalPlans.scala → sql/hbase/src/main/scala/org/apache/spark/sql/hbase/logical/hbaseOperators.scala

@@ -16,7 +16,7 @@
  */
 package org.apache.spark.sql.hbase.logical
 
-import org.apache.spark.sql.catalyst.plans.logical.{LeafNode, Command}
+import org.apache.spark.sql.catalyst.plans.logical.{UnaryNode, LeafNode, LogicalPlan, Command}
 import org.apache.spark.sql.hbase.HBaseRelation
 
 case class CreateHBaseTablePlan(tableName: String,
@@ -29,6 +29,17 @@ case class CreateHBaseTablePlan(tableName: String,
 
 case class DropTablePlan(tableName: String) extends Command
 
-case class LoadDataIntoTable(path: String, table: String, isLocal: Boolean) extends LeafNode {
-  override def output = Seq.empty
+/**
+ * Logical plan for Bulkload
+ * @param path input data file path
+ * @param child target relation
+ * @param isLocal using HDFS or local file
+ */
+case class LoadDataIntoTablePlan(path: String,
+                             child: LogicalPlan,
+                             isLocal: Boolean) extends UnaryNode {
+
+  override def output = Nil
+
+  override def toString = s"LogicalPlan: LoadDataIntoTable(LOAD $path INTO $child)"
 }

sql/hbase/src/test/scala/org/apache/spark/sql/hbase/HBaseBulkloadParserTest.scala

@@ -0,0 +1,70 @@
+/*
+ * 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.hbase
+
+import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.hbase.logical.LoadDataIntoTablePlan
+import org.scalatest.{FunSuite, Matchers}
+
+/**
+ * Test Suite for HBase bulkload feature
+ * Created by jackylk on 2014/10/25.
+ */
+
+class HBaseBulkloadParserTest extends FunSuite with Matchers {
+
+  // Test if we can parse 'LOAD DATA LOCAL INPATH './usr/file.csv' INTO TABLE tb'
+  test("bulkload parser test, local file") {
+
+    val parser = new HBaseSQLParser()
+    val sql = raw"LOAD DATA LOCAL INPATH './usr/file.csv' INTO TABLE tb"
+    //val sql = "select"
+
+    val plan: LogicalPlan = parser(sql)
+    assert(plan != null)
+    assert(plan.isInstanceOf[LoadDataIntoTablePlan])
+
+    val l = plan.asInstanceOf[LoadDataIntoTablePlan]
+    assert(l.path.equals(raw"./usr/file.csv"))
+    assert(l.isLocal)
+
+    assert(plan.children(0).isInstanceOf[UnresolvedRelation])
+    val r = plan.children(0).asInstanceOf[UnresolvedRelation]
+    assert(r.tableName.equals("tb"))
+  }
+
+  // Test if we can parse 'LOAD DATA INPATH '/usr/hdfsfile.csv' INTO TABLE tb'
+  test("bulkload parser test, load hdfs file") {
+
+    val parser = new HBaseSQLParser()
+    val sql = raw"LOAD DATA INPATH '/usr/hdfsfile.csv' INTO TABLE tb"
+    //val sql = "select"
+
+    val plan: LogicalPlan = parser(sql)
+    assert(plan != null)
+    assert(plan.isInstanceOf[LoadDataIntoTablePlan])
+
+    val l = plan.asInstanceOf[LoadDataIntoTablePlan]
+    assert(l.path.equals(raw"/usr/hdfsfile.csv"))
+    assert(!l.isLocal)
+    assert(plan.children(0).isInstanceOf[UnresolvedRelation])
+    val r = plan.children(0).asInstanceOf[UnresolvedRelation]
+    assert(r.tableName.equals("tb"))
+  }
+}

sql/hbase/src/test/scala/org/apache/spark/sql/hbase/HBasePartitionerSuite.scala

@@ -19,7 +19,7 @@ package org.apache.spark.sql.hbase
 
 import org.scalatest.FunSuite
 import org.apache.spark.{SparkConf, LocalSparkContext, SparkContext, Logging}
-
+import org.apache.spark.SparkContext._
 class HBasePartitionerSuite extends FunSuite with LocalSparkContext with Logging {
 
   val conf = new SparkConf(loadDefaults = false)