Fixed dataframe api

apache · Jul 8, 2016 · 60a9d2f · 60a9d2f
1 parent 8dfa672
commit 60a9d2f
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 15 deletions.
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/CarbonDatasourceRelation.scala b/integration/spark/src/main/scala/org/apache/spark/sql/CarbonDatasourceRelation.scala
@@ -52,20 +52,19 @@ class CarbonSource
   override def createRelation(
       sqlContext: SQLContext,
       parameters: Map[String, String]): BaseRelation = {
-    if (parameters.get("tablePath") != None) {
-      val options = new CarbonOption(parameters)
-      val tableIdentifier = options.tableIdentifier.split("""\.""").toSeq
-      val ident = tableIdentifier match {
-        case Seq(name) => TableIdentifier(name, None)
-        case Seq(db, name) => TableIdentifier(name, Some(db))
-      }
-      CarbonDatasourceRelation(ident, None)(sqlContext)
-    } else if (parameters.get("path") != None) {
-      CarbonDatasourceHadoopRelation(sqlContext, Array(parameters.get("path").get), parameters)
-    } else {
-      sys.error("Carbon table path not found")
+    // if path is provided we can directly create Hadoop relation. \
+    // Otherwise create datasource relation
+    parameters.get("path") match {
+      case Some(path) => CarbonDatasourceHadoopRelation(sqlContext, Array(path), parameters)
+      case _ =>
+        val options = new CarbonOption(parameters)
+        val tableIdentifier = options.tableIdentifier.split("""\.""").toSeq
+        val identifier = tableIdentifier match {
+          case Seq(name) => TableIdentifier(name, None)
+          case Seq(db, name) => TableIdentifier(name, Some(db))
+        }
+        CarbonDatasourceRelation(identifier, None)(sqlContext)
     }
-
   }
 
   override def createRelation(
@@ -93,7 +92,7 @@ class CarbonSource
         sys.error(s"ErrorIfExists mode, path $storePath already exists.")
       case (SaveMode.Overwrite, true) =>
         val cc = CarbonContext.getInstance(sqlContext.sparkContext)
-        cc.sql(s"DROP CUBE IF EXISTS ${ options.dbName }.${ options.tableName }")
+        cc.sql(s"DROP TABLE IF EXISTS ${ options.dbName }.${ options.tableName }")
         (true, false)
       case (SaveMode.Overwrite, false) | (SaveMode.ErrorIfExists, false) =>
         (true, false)

diff --git a/integration/spark/src/main/scala/org/carbondata/spark/package.scala b/integration/spark/src/main/scala/org/carbondata/spark/package.scala
@@ -63,7 +63,7 @@ package object spark {
           if (f.getPath.getName.startsWith("part-")) {
             val newPath = s"${ f.getPath.getParent }/${ f.getPath.getName }.csv"
             if (!fs.rename(f.getPath, new Path(newPath))) {
-              cc.sql(s"DROP CUBE ${ options.tableName }")
+              cc.sql(s"DROP TABLE ${ options.tableName }")
               throw new RuntimeException("File system rename failed when loading data into carbon")
             }
           }