You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ja...@apache.org on 2017/01/16 12:14:15 UTC
[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
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 <ra...@gmail.com>
Authored: Mon Jan 9 18:58:13 2017 +0530
Committer: jackylk <ja...@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()
[2/2] incubator-carbondata git commit: [CARBONDATA-440] fixing no
kettle issue for IUD. This closes #523
Posted by ja...@apache.org.
[CARBONDATA-440] fixing no kettle issue for IUD. This closes #523
Project: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/commit/1d5b5fcc
Tree: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/tree/1d5b5fcc
Diff: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/diff/1d5b5fcc
Branch: refs/heads/master
Commit: 1d5b5fcc4f792db9eafd3318b4ef96df9aa04e82
Parents: b5c20a8 2cf2d19
Author: jackylk <ja...@huawei.com>
Authored: Mon Jan 16 20:14:04 2017 +0800
Committer: jackylk <ja...@huawei.com>
Committed: Mon Jan 16 20:14:04 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(-)
----------------------------------------------------------------------