carbondata-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jack...@apache.org
Subject [1/2] incubator-carbondata git commit: fixing no kettle issue for IUD. load count/ segment count should be string because in compaction case it will be 2.1
Date Mon, 16 Jan 2017 12:14:15 GMT
Repository: incubator-carbondata
Updated Branches:
  refs/heads/master b5c20a80a -> 1d5b5fcc4


fixing no kettle issue for IUD.
load count/ segment count should be string because in compaction case it will be 2.1


Project: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/commit/2cf2d193
Tree: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/tree/2cf2d193
Diff: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/diff/2cf2d193

Branch: refs/heads/master
Commit: 2cf2d1936237144a23c0a4393584a15b7beb600c
Parents: b5c20a8
Author: ravikiran <ravikiran.sn042@gmail.com>
Authored: Mon Jan 9 18:58:13 2017 +0530
Committer: jackylk <jacky.likun@huawei.com>
Committed: Mon Jan 16 20:13:24 2017 +0800

----------------------------------------------------------------------
 .../spark/rdd/CarbonDataLoadRDD.scala           |  6 +-
 .../spark/rdd/NewCarbonDataLoadRDD.scala        |  4 +-
 .../carbondata/spark/rdd/UpdateDataLoad.scala   | 79 ++++++++++++++++++++
 .../spark/rdd/CarbonDataRDDFactory.scala        | 49 +++++++-----
 .../iud/UpdateCarbonTableTestCase.scala         | 13 ++++
 5 files changed, 128 insertions(+), 23 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/2cf2d193/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataLoadRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataLoadRDD.scala
b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataLoadRDD.scala
index ff3a174..c8e3c67 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataLoadRDD.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataLoadRDD.scala
@@ -83,7 +83,7 @@ class SparkPartitionLoader(model: CarbonLoadModel,
     splitIndex: Int,
     storePath: String,
     kettleHomePath: String,
-    loadCount: Int,
+    loadCount: String,
     loadMetadataDetails: LoadMetadataDetails) {
   private val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
 
@@ -239,7 +239,7 @@ class DataFileLoaderRDD[K, V](
         carbonLoadModel.setSegmentId(String.valueOf(loadCount))
         setModelAndBlocksInfo()
         val loader = new SparkPartitionLoader(model, theSplit.index, storePath,
-          kettleHomePath, loadCount, loadMetadataDetails)
+          kettleHomePath, String.valueOf(loadCount), loadMetadataDetails)
         loader.initialize
         if (model.isRetentionRequest) {
           recreateAggregationTableForRetention
@@ -511,7 +511,7 @@ class DataFrameLoaderRDD[K, V](
         carbonLoadModel.setSegmentId(String.valueOf(loadCount))
         carbonLoadModel.setTaskNo(String.valueOf(theSplit.index))
         val loader = new SparkPartitionLoader(carbonLoadModel, theSplit.index, storePath,
-          kettleHomePath, loadCount, loadMetadataDetails)
+          kettleHomePath, String.valueOf(loadCount), loadMetadataDetails)
         loader.initialize
         val rddIteratorKey = UUID.randomUUID().toString
         try {

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/2cf2d193/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/NewCarbonDataLoadRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/NewCarbonDataLoadRDD.scala
b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/NewCarbonDataLoadRDD.scala
index 46e83a5..9e13883 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/NewCarbonDataLoadRDD.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/NewCarbonDataLoadRDD.scala
@@ -157,7 +157,7 @@ class NewCarbonDataLoadRDD[K, V](
           theSplit.index,
           null,
           null,
-          loadCount,
+          String.valueOf(loadCount),
           loadMetadataDetails)
         // Intialize to set carbon properties
         loader.initialize()
@@ -355,7 +355,7 @@ class NewDataFrameLoaderRDD[K, V](
           theSplit.index,
           null,
           null,
-          loadCount,
+          String.valueOf(loadCount),
           loadMetadataDetails)
         // Intialize to set carbon properties
         loader.initialize()

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/2cf2d193/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/UpdateDataLoad.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/UpdateDataLoad.scala
b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/UpdateDataLoad.scala
new file mode 100644
index 0000000..7dee88a
--- /dev/null
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/UpdateDataLoad.scala
@@ -0,0 +1,79 @@
+/*
+ * 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.carbondata.spark.rdd
+
+import java.nio.ByteBuffer
+
+import scala.collection.mutable
+
+import org.apache.spark.{SparkEnv, TaskContext}
+import org.apache.spark.sql.Row
+
+import org.apache.carbondata.common.CarbonIterator
+import org.apache.carbondata.common.logging.LogServiceFactory
+import org.apache.carbondata.core.constants.CarbonCommonConstants
+import org.apache.carbondata.core.load.LoadMetadataDetails
+import org.apache.carbondata.processing.model.CarbonLoadModel
+import org.apache.carbondata.processing.newflow.DataLoadExecutor
+import org.apache.carbondata.processing.newflow.exception.BadRecordFoundException
+
+/**
+ * Data load in case of update command with out kettle.
+ */
+object UpdateDataLoad {
+
+  def DataLoadNoKettleForUpdate(segId: String,
+      index: Int,
+      iter: Iterator[Row],
+      carbonLoadModel: CarbonLoadModel,
+      loadMetadataDetails: LoadMetadataDetails): Unit = {
+    val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
+    try {
+      val recordReaders = mutable.Buffer[CarbonIterator[Array[AnyRef]]]()
+      val serializer = SparkEnv.get.closureSerializer.newInstance()
+      var serializeBuffer: ByteBuffer = null
+      recordReaders += new NewRddIterator(iter,
+          carbonLoadModel,
+          TaskContext.get())
+
+      val loader = new SparkPartitionLoader(carbonLoadModel,
+        index,
+        null,
+        null,
+        segId,
+        loadMetadataDetails)
+      // Intialize to set carbon properties
+      loader.initialize()
+
+      loadMetadataDetails.setLoadStatus(CarbonCommonConstants.STORE_LOADSTATUS_SUCCESS)
+      new DataLoadExecutor().execute(carbonLoadModel,
+        loader.storeLocation,
+        recordReaders.toArray)
+
+    } catch {
+      case e: BadRecordFoundException =>
+        loadMetadataDetails.setLoadStatus(
+          CarbonCommonConstants.STORE_LOADSTATUS_PARTIAL_SUCCESS)
+        LOGGER.info("Bad Record Found")
+      case e: Exception =>
+        LOGGER.error(e)
+        throw e
+    }
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/2cf2d193/integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
b/integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
index 7a7aa64..610f5fb 100644
--- a/integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
+++ b/integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
@@ -17,11 +17,13 @@
 
 package org.apache.carbondata.spark.rdd
 
+import java.nio.ByteBuffer
 import java.util
 import java.util.UUID
 import java.util.concurrent._
 
 import scala.collection.JavaConverters._
+import scala.collection.mutable
 import scala.collection.mutable.ListBuffer
 import scala.util.Random
 import scala.util.control.Breaks._
@@ -30,36 +32,35 @@ import org.apache.hadoop.conf.{Configurable, Configuration}
 import org.apache.hadoop.fs.Path
 import org.apache.hadoop.mapreduce.Job
 import org.apache.hadoop.mapreduce.lib.input.{FileInputFormat, FileSplit}
-import org.apache.spark.{SparkContext, SparkEnv, SparkException}
+import org.apache.spark.{SparkContext, SparkEnv, SparkException, TaskContext}
 import org.apache.spark.rdd.{DataLoadCoalescedRDD, DataLoadPartitionCoalescer, UpdateCoalescedRDD}
 import org.apache.spark.sql.{CarbonEnv, DataFrame, Row, SQLContext}
-import org.apache.spark.sql.execution.command.{AlterTableModel, CompactionCallableModel,
CompactionModel, ExecutionErrors, UpdateTableModel}
+import org.apache.spark.sql.execution.command.{AlterTableModel, CompactionModel, ExecutionErrors,
UpdateTableModel}
 import org.apache.spark.sql.hive.DistributionUtil
 import org.apache.spark.util.SparkUtil
 
+import org.apache.carbondata.common.CarbonIterator
 import org.apache.carbondata.common.logging.LogServiceFactory
-import org.apache.carbondata.core.carbon.{CarbonDataLoadSchema, CarbonTableIdentifier, ColumnarFormatVersion}
+import org.apache.carbondata.core.carbon.{CarbonTableIdentifier, ColumnarFormatVersion}
 import org.apache.carbondata.core.carbon.datastore.block.{Distributable, TableBlockInfo}
-import org.apache.carbondata.core.carbon.metadata.CarbonMetadata
 import org.apache.carbondata.core.carbon.metadata.schema.table.CarbonTable
 import org.apache.carbondata.core.carbon.path.CarbonStorePath
 import org.apache.carbondata.core.constants.CarbonCommonConstants
 import org.apache.carbondata.core.dictionary.server.DictionaryServer
 import org.apache.carbondata.core.load.{BlockDetails, LoadMetadataDetails}
 import org.apache.carbondata.core.update.CarbonUpdateUtil
-import org.apache.carbondata.core.updatestatus.SegmentStatusManager
 import org.apache.carbondata.core.util.CarbonProperties
 import org.apache.carbondata.locks.{CarbonLockFactory, ICarbonLock, LockUsage}
 import org.apache.carbondata.processing.csvreaderstep.RddInpututilsForUpdate
 import org.apache.carbondata.processing.etl.DataLoadingException
 import org.apache.carbondata.processing.model.CarbonLoadModel
-import org.apache.carbondata.processing.newflow.exception.CarbonDataLoadingException
+import org.apache.carbondata.processing.newflow.DataLoadExecutor
+import org.apache.carbondata.processing.newflow.exception.{BadRecordFoundException, CarbonDataLoadingException}
 import org.apache.carbondata.spark._
 import org.apache.carbondata.spark.load._
-import org.apache.carbondata.spark.merger.{CarbonCompactionUtil, CarbonDataMergerUtil, CompactionCallable,
CompactionType}
-import org.apache.carbondata.spark.partition.api.Partition
+import org.apache.carbondata.spark.merger.{CarbonCompactionUtil, CarbonDataMergerUtil, CompactionType}
 import org.apache.carbondata.spark.splits.TableSplit
-import org.apache.carbondata.spark.util.{CarbonQueryUtil, CommonUtil, LoadMetadataUtil}
+import org.apache.carbondata.spark.util.{CarbonQueryUtil, CommonUtil}
 
 /**
  * This is the factory class which can create different RDD depends on user needs.
@@ -715,16 +716,28 @@ object CarbonDataRDDFactory {
               loadMetadataDetails.setLoadStatus(CarbonCommonConstants.STORE_LOADSTATUS_SUCCESS)
               val rddIteratorKey = CarbonCommonConstants.RDDUTIL_UPDATE_KEY +
                                    UUID.randomUUID().toString
+              if (useKettle) {
+                try {
+                  RddInpututilsForUpdate.put(rddIteratorKey,
+                    new RddIteratorForUpdate(iter, carbonLoadModel))
+                  carbonLoadModel.setRddIteratorKey(rddIteratorKey)
+                  CarbonDataLoadForUpdate.run(carbonLoadModel,
+                    index,
+                    storePath,
+                    kettleHomePath,
+                    segId,
+                    loadMetadataDetails,
+                    executionErrors)
+                } finally {
+                  RddInpututilsForUpdate.remove(rddIteratorKey)
+                }
+              } else {
+                UpdateDataLoad.DataLoadNoKettleForUpdate(segId,
+                  index,
+                  iter,
+                  carbonLoadModel,
+                  loadMetadataDetails)
 
-              try {
-                RddInpututilsForUpdate.put(rddIteratorKey,
-                  new RddIteratorForUpdate(iter, carbonLoadModel))
-                carbonLoadModel.setRddIteratorKey(rddIteratorKey)
-                CarbonDataLoadForUpdate
-                  .run(carbonLoadModel, index, storePath, kettleHomePath,
-                    segId, loadMetadataDetails, executionErrors)
-              } finally {
-                RddInpututilsForUpdate.remove(rddIteratorKey)
               }
             } catch {
               case e: Exception =>

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/2cf2d193/integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/iud/UpdateCarbonTableTestCase.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/iud/UpdateCarbonTableTestCase.scala
b/integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/iud/UpdateCarbonTableTestCase.scala
index b4c6940..2eb859b 100644
--- a/integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/iud/UpdateCarbonTableTestCase.scala
+++ b/integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/iud/UpdateCarbonTableTestCase.scala
@@ -102,6 +102,19 @@ class UpdateCarbonTableTestCase extends QueryTest with BeforeAndAfterAll
{
       sql("""drop table iud.dest33""").show
   }
 
+  test("update carbon table without alias in set columns with mulitple loads") {
+    sql("""drop table iud.dest33""").show
+    sql("""create table iud.dest33 (c1 string,c2 int,c3 string,c5 string) STORED BY 'org.apache.carbondata.format'""").show()
+    sql(s"""LOAD DATA LOCAL INPATH '$resourcesPath/IUD/dest.csv' INTO table iud.dest33""")
+    sql(s"""LOAD DATA LOCAL INPATH '$resourcesPath/IUD/dest.csv' INTO table iud.dest33""")
+    sql("""update iud.dest33 d set (c3,c5 ) = (select s.c33 ,s.c55  from iud.source2 s where
d.c1 = s.c11) where d.c1 = 'a'""").show()
+    checkAnswer(
+      sql("""select c3,c5 from iud.dest33 where c1='a'"""),
+      Seq(Row("MGM","Disco"),Row("MGM","Disco"))
+    )
+    sql("""drop table iud.dest33""").show
+  }
+
    test("update carbon table without alias in set three columns") {
      sql("""drop table iud.dest44""").show
      sql("""create table iud.dest44 (c1 string,c2 int,c3 string,c5 string) STORED BY 'org.apache.carbondata.format'""").show()


Mime
View raw message