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 2018/01/13 08:10:30 UTC
[01/14] carbondata git commit: [CARBONDATA-1999] Block drop table and
delete streaming segment while streaming is in progress
Repository: carbondata
Updated Branches:
refs/heads/carbonstore 90921eb7b -> b8bcb0e0e
[CARBONDATA-1999] Block drop table and delete streaming segment while streaming is in progress
1.Block drop table while streaming is in progress
2.Block delete streaming segment
This closes #1773
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/1b72a02b
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/1b72a02b
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/1b72a02b
Branch: refs/heads/carbonstore
Commit: 1b72a02be2f176bcc99dc12f7621a0623d39b973
Parents: 90921eb
Author: QiangCai <qi...@qq.com>
Authored: Mon Jan 8 11:14:00 2018 +0800
Committer: Jacky Li <ja...@qq.com>
Committed: Tue Jan 9 00:49:27 2018 +0800
----------------------------------------------------------------------
.../statusmanager/SegmentStatusManager.java | 42 ++++++-----
.../command/table/CarbonDropTableCommand.scala | 4 +
.../sql/execution/strategy/DDLStrategy.scala | 2 +-
.../TestStreamingTableOperation.scala | 77 ++++++++++++++------
4 files changed, 85 insertions(+), 40 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/1b72a02b/core/src/main/java/org/apache/carbondata/core/statusmanager/SegmentStatusManager.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/statusmanager/SegmentStatusManager.java b/core/src/main/java/org/apache/carbondata/core/statusmanager/SegmentStatusManager.java
index 7804ea8..e1fadcf 100755
--- a/core/src/main/java/org/apache/carbondata/core/statusmanager/SegmentStatusManager.java
+++ b/core/src/main/java/org/apache/carbondata/core/statusmanager/SegmentStatusManager.java
@@ -473,6 +473,7 @@ public class SegmentStatusManager {
*/
private static List<String> updateDeletionStatus(List<String> loadIds,
LoadMetadataDetails[] listOfLoadFolderDetailsArray, List<String> invalidLoadIds) {
+ SegmentStatus segmentStatus = null;
for (String loadId : loadIds) {
boolean loadFound = false;
// For each load id loop through data and if the
@@ -481,26 +482,29 @@ public class SegmentStatusManager {
for (LoadMetadataDetails loadMetadata : listOfLoadFolderDetailsArray) {
if (loadId.equalsIgnoreCase(loadMetadata.getLoadName())) {
- // if the segment is compacted then no need to delete that.
- if (SegmentStatus.COMPACTED == loadMetadata.getSegmentStatus()) {
+ segmentStatus = loadMetadata.getSegmentStatus();
+ if (SegmentStatus.COMPACTED == segmentStatus) {
+ // if the segment is compacted then no need to delete that.
LOG.error("Cannot delete the Segment which is compacted. Segment is " + loadId);
invalidLoadIds.add(loadId);
return invalidLoadIds;
- }
- // if the segment status is in progress then no need to delete that.
- if (SegmentStatus.INSERT_IN_PROGRESS == loadMetadata.getSegmentStatus()) {
+ } else if (SegmentStatus.INSERT_IN_PROGRESS == segmentStatus) {
+ // if the segment status is in progress then no need to delete that.
LOG.error("Cannot delete the segment " + loadId + " which is load in progress");
invalidLoadIds.add(loadId);
return invalidLoadIds;
- }
- // if the segment status is overwrite in progress, then no need to delete that.
- if (SegmentStatus.INSERT_OVERWRITE_IN_PROGRESS == loadMetadata.getSegmentStatus()) {
- LOG.error("Cannot delete the segemnt " + loadId + " which is load overwrite " +
+ } else if (SegmentStatus.INSERT_OVERWRITE_IN_PROGRESS == segmentStatus) {
+ // if the segment status is overwrite in progress, then no need to delete that.
+ LOG.error("Cannot delete the segment " + loadId + " which is load overwrite " +
"in progress");
invalidLoadIds.add(loadId);
return invalidLoadIds;
- }
- if (SegmentStatus.MARKED_FOR_DELETE != loadMetadata.getSegmentStatus()) {
+ } else if (SegmentStatus.STREAMING == segmentStatus) {
+ // if the segment status is streaming, the segment can't be deleted directly.
+ LOG.error("Cannot delete the segment " + loadId + " which is streaming in progress");
+ invalidLoadIds.add(loadId);
+ return invalidLoadIds;
+ } else if (SegmentStatus.MARKED_FOR_DELETE != segmentStatus) {
loadFound = true;
loadMetadata.setSegmentStatus(SegmentStatus.MARKED_FOR_DELETE);
loadMetadata.setModificationOrdeletionTimesStamp(CarbonUpdateUtil.readCurrentTime());
@@ -535,17 +539,20 @@ public class SegmentStatusManager {
// the metadata as deleted.
boolean loadFound = false;
String loadStartTimeString = "Load Start Time: ";
+ SegmentStatus segmentStatus = null;
for (LoadMetadataDetails loadMetadata : listOfLoadFolderDetailsArray) {
Integer result = compareDateValues(loadMetadata.getLoadStartTimeAsLong(), loadStartTime);
if (result < 0) {
- if (SegmentStatus.COMPACTED == loadMetadata.getSegmentStatus()) {
+ segmentStatus = loadMetadata.getSegmentStatus();
+ if (SegmentStatus.COMPACTED == segmentStatus) {
LOG.info("Ignoring the segment : " + loadMetadata.getLoadName()
+ "as the segment has been compacted.");
- continue;
- }
- if (SegmentStatus.MARKED_FOR_DELETE != loadMetadata.getSegmentStatus() &&
- SegmentStatus.INSERT_IN_PROGRESS != loadMetadata.getSegmentStatus() &&
- SegmentStatus.INSERT_OVERWRITE_IN_PROGRESS != loadMetadata.getSegmentStatus()) {
+ } else if (SegmentStatus.STREAMING == segmentStatus) {
+ LOG.info("Ignoring the segment : " + loadMetadata.getLoadName()
+ + "as the segment is streaming in progress.");
+ } else if (SegmentStatus.MARKED_FOR_DELETE != segmentStatus &&
+ SegmentStatus.INSERT_IN_PROGRESS != segmentStatus &&
+ SegmentStatus.INSERT_OVERWRITE_IN_PROGRESS != segmentStatus) {
loadFound = true;
updateSegmentMetadataDetails(loadMetadata);
LOG.info("Info: " +
@@ -553,7 +560,6 @@ public class SegmentStatusManager {
" Marked for Delete");
}
}
-
}
if (!loadFound) {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/1b72a02b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonDropTableCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonDropTableCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonDropTableCommand.scala
index 9901f8c..aaad207 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonDropTableCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonDropTableCommand.scala
@@ -55,6 +55,10 @@ case class CarbonDropTableCommand(
}
LOGGER.audit(s"Deleting table [$tableName] under database [$dbName]")
carbonTable = CarbonEnv.getCarbonTable(databaseNameOp, tableName)(sparkSession)
+ if (carbonTable.isStreamingTable) {
+ // streaming table should acquire streaming.lock
+ carbonLocks += CarbonLockUtil.getLockObject(identifier, LockUsage.STREAMING_LOCK)
+ }
val relationIdentifiers = carbonTable.getTableInfo.getParentRelationIdentifiers
if (relationIdentifiers != null && !relationIdentifiers.isEmpty) {
if (!dropChildTable) {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/1b72a02b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/strategy/DDLStrategy.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/strategy/DDLStrategy.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/strategy/DDLStrategy.scala
index 2805114..6ff762a 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/strategy/DDLStrategy.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/strategy/DDLStrategy.scala
@@ -223,7 +223,7 @@ class DDLStrategy(sparkSession: SparkSession) extends SparkStrategy {
if (property.isDefined) {
if (!property.get._2.trim.equalsIgnoreCase("true")) {
throw new MalformedCarbonCommandException(
- "Streaming property can not be changed to 'false' once it is 'true'")
+ "Streaming property can not be changed once it is 'true'")
}
}
ExecutedCommandExec(CarbonAlterTableSetCommand(tableName, properties, isView)) :: Nil
http://git-wip-us.apache.org/repos/asf/carbondata/blob/1b72a02b/integration/spark2/src/test/scala/org/apache/spark/carbondata/TestStreamingTableOperation.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/test/scala/org/apache/spark/carbondata/TestStreamingTableOperation.scala b/integration/spark2/src/test/scala/org/apache/spark/carbondata/TestStreamingTableOperation.scala
index 62ab9af..a8ab6fb 100644
--- a/integration/spark2/src/test/scala/org/apache/spark/carbondata/TestStreamingTableOperation.scala
+++ b/integration/spark2/src/test/scala/org/apache/spark/carbondata/TestStreamingTableOperation.scala
@@ -32,7 +32,6 @@ import org.apache.spark.sql.types.StructType
import org.scalatest.BeforeAndAfterAll
import org.apache.carbondata.core.constants.CarbonCommonConstants
-import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.statusmanager.{FileFormat, SegmentStatus}
import org.apache.carbondata.core.util.path.{CarbonStorePath, CarbonTablePath}
import org.apache.carbondata.spark.exception.MalformedCarbonCommandException
@@ -105,7 +104,8 @@ class TestStreamingTableOperation extends QueryTest with BeforeAndAfterAll {
createTable(tableName = "stream_table_tolerant", streaming = true, withBatchLoad = true)
// 11. table for delete segment test
- createTable(tableName = "stream_table_delete", streaming = true, withBatchLoad = false)
+ createTable(tableName = "stream_table_delete_id", streaming = true, withBatchLoad = false)
+ createTable(tableName = "stream_table_delete_date", streaming = true, withBatchLoad = false)
// 12. reject alter streaming properties
createTable(tableName = "stream_table_alter", streaming = false, withBatchLoad = false)
@@ -126,6 +126,8 @@ class TestStreamingTableOperation extends QueryTest with BeforeAndAfterAll {
// 17. reopen streaming table after close
createTable(tableName = "stream_table_reopen", streaming = true, withBatchLoad = false)
+ // 18. block drop table while streaming is in progress
+ createTable(tableName = "stream_table_drop", streaming = true, withBatchLoad = false)
}
test("validate streaming property") {
@@ -204,7 +206,8 @@ class TestStreamingTableOperation extends QueryTest with BeforeAndAfterAll {
sql("drop table if exists streaming.stream_table_compact")
sql("drop table if exists streaming.stream_table_new")
sql("drop table if exists streaming.stream_table_tolerant")
- sql("drop table if exists streaming.stream_table_delete")
+ sql("drop table if exists streaming.stream_table_delete_id")
+ sql("drop table if exists streaming.stream_table_delete_date")
sql("drop table if exists streaming.stream_table_alter")
sql("drop table if exists streaming.stream_table_handoff")
sql("drop table if exists streaming.stream_table_finish")
@@ -212,6 +215,7 @@ class TestStreamingTableOperation extends QueryTest with BeforeAndAfterAll {
sql("drop table if exists streaming.stream_table_close")
sql("drop table if exists streaming.stream_table_close_auto_handoff")
sql("drop table if exists streaming.stream_table_reopen")
+ sql("drop table if exists streaming.stream_table_drop")
}
// normal table not support streaming ingest
@@ -600,7 +604,6 @@ class TestStreamingTableOperation extends QueryTest with BeforeAndAfterAll {
}
sql("alter table streaming.stream_table_compact compact 'minor'")
- sql("show segments for table streaming.stream_table_compact").show
val result = sql("show segments for table streaming.stream_table_compact").collect()
result.foreach { row =>
@@ -635,48 +638,55 @@ class TestStreamingTableOperation extends QueryTest with BeforeAndAfterAll {
test("test deleting streaming segment by ID while ingesting") {
executeStreamingIngest(
- tableName = "stream_table_delete",
+ tableName = "stream_table_delete_id",
batchNums = 6,
rowNumsEachBatch = 10000,
intervalOfSource = 3,
intervalOfIngest = 5,
- continueSeconds = 15,
+ continueSeconds = 20,
generateBadRecords = false,
badRecordAction = "force",
handoffSize = 1024L * 200,
autoHandoff = false
)
- val beforeDelete = sql("show segments for table streaming.stream_table_delete").collect()
- val segmentId = beforeDelete.map(_.getString(0)).mkString(",")
- sql(s"delete from table streaming.stream_table_delete where segment.id in ($segmentId) ")
+ Thread.sleep(10000)
+ val beforeDelete = sql("show segments for table streaming.stream_table_delete_id").collect()
+ val segmentIds1 = beforeDelete.filter(_.getString(1).equals("Streaming")).map(_.getString(0)).mkString(",")
+ val msg = intercept[Exception] {
+ sql(s"delete from table streaming.stream_table_delete_id where segment.id in ($segmentIds1) ")
+ }
+ assertResult(s"Delete segment by Id is failed. Invalid ID is: ${beforeDelete.length -1}")(msg.getMessage)
- val rows = sql("show segments for table streaming.stream_table_delete").collect()
- rows.foreach { row =>
+ val segmentIds2 = beforeDelete.filter(_.getString(1).equals("Streaming Finish"))
+ .map(_.getString(0)).mkString(",")
+ sql(s"delete from table streaming.stream_table_delete_id where segment.id in ($segmentIds2) ")
+ val afterDelete = sql("show segments for table streaming.stream_table_delete_id").collect()
+ afterDelete.filter(!_.getString(1).equals("Streaming")).foreach { row =>
assertResult(SegmentStatus.MARKED_FOR_DELETE.getMessage)(row.getString(1))
}
}
test("test deleting streaming segment by date while ingesting") {
executeStreamingIngest(
- tableName = "stream_table_delete",
+ tableName = "stream_table_delete_date",
batchNums = 6,
rowNumsEachBatch = 10000,
intervalOfSource = 3,
intervalOfIngest = 5,
- continueSeconds = 15,
+ continueSeconds = 20,
generateBadRecords = false,
badRecordAction = "force",
handoffSize = 1024L * 200,
autoHandoff = false
)
- val beforeDelete = sql("show segments for table streaming.stream_table_delete").collect()
-
- sql(s"delete from table streaming.stream_table_delete where segment.starttime before " +
+ Thread.sleep(10000)
+ val beforeDelete = sql("show segments for table streaming.stream_table_delete_date").collect()
+ sql(s"delete from table streaming.stream_table_delete_date where segment.starttime before " +
s"'2999-10-01 01:00:00'")
-
- val rows = sql("show segments for table streaming.stream_table_delete").collect()
- assertResult(beforeDelete.length)(rows.length)
- rows.foreach { row =>
+ val segmentIds = beforeDelete.filter(_.getString(1).equals("Streaming"))
+ assertResult(1)(segmentIds.length)
+ val afterDelete = sql("show segments for table streaming.stream_table_delete_date").collect()
+ afterDelete.filter(!_.getString(1).equals("Streaming")).foreach { row =>
assertResult(SegmentStatus.MARKED_FOR_DELETE.getMessage)(row.getString(1))
}
}
@@ -799,7 +809,6 @@ class TestStreamingTableOperation extends QueryTest with BeforeAndAfterAll {
autoHandoff = false
)
sql("alter table streaming.stream_table_finish finish streaming")
- sql("show segments for table streaming.stream_table_finish").show(100, false)
val segments = sql("show segments for table streaming.stream_table_finish").collect()
assert(segments.length == 4 || segments.length == 5)
@@ -938,6 +947,32 @@ class TestStreamingTableOperation extends QueryTest with BeforeAndAfterAll {
)
}
+ test("block drop streaming table while streaming is in progress") {
+ val identifier = new TableIdentifier("stream_table_drop", Option("streaming"))
+ val carbonTable = CarbonEnv.getInstance(spark).carbonMetastore.lookupRelation(identifier)(spark)
+ .asInstanceOf[CarbonRelation].metaData.carbonTable
+ val tablePath = CarbonStorePath.getCarbonTablePath(carbonTable.getAbsoluteTableIdentifier)
+ var server: ServerSocket = null
+ try {
+ server = getServerSocket
+ val thread1 = createWriteSocketThread(server, 2, 10, 5)
+ val thread2 = createSocketStreamingThread(spark, server.getLocalPort, tablePath, identifier, "force", 5, 1024L * 200, false)
+ thread1.start()
+ thread2.start()
+ Thread.sleep(1000)
+ val msg = intercept[Exception] {
+ sql(s"drop table streaming.stream_table_drop")
+ }
+ assertResult("Dropping table streaming.stream_table_drop failed: Acquire table lock failed after retry, please try after some time;")(msg.getMessage)
+ thread1.interrupt()
+ thread2.interrupt()
+ } finally {
+ if (server != null) {
+ server.close()
+ }
+ }
+ }
+
test("do not support creating datamap on streaming table") {
assert(
intercept[MalformedCarbonCommandException](
[03/14] carbondata git commit: [CARBONDATA-1755] Fixed bug occuring
on concurrent insert-overwrite and update
Posted by ja...@apache.org.
[CARBONDATA-1755] Fixed bug occuring on concurrent insert-overwrite and update
Description: Concurrent Insert overwrite-update: User is able to run insert overwrite and update job concurrently.
Updated data will be overwritten by insert overwrite job. So there is no meaning of running update job if insert overwrite is in progress. So the two operations should not be allowed to run at the same time.
This closes #1722
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/62ce5b5a
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/62ce5b5a
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/62ce5b5a
Branch: refs/heads/carbonstore
Commit: 62ce5b5abaffce13d3e2a169e80c361a126ab65f
Parents: d33d347
Author: SangeetaGulia <sa...@knoldus.in>
Authored: Fri Dec 22 16:38:10 2017 +0530
Committer: Jacky Li <ja...@qq.com>
Committed: Tue Jan 9 21:46:12 2018 +0800
----------------------------------------------------------------------
.../iud/TestInsertUpdateConcurrentTest.scala | 101 +++++++++++++++++++
.../CarbonProjectForUpdateCommand.scala | 9 ++
2 files changed, 110 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/62ce5b5a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/TestInsertUpdateConcurrentTest.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/TestInsertUpdateConcurrentTest.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/TestInsertUpdateConcurrentTest.scala
new file mode 100644
index 0000000..613b2a6
--- /dev/null
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/TestInsertUpdateConcurrentTest.scala
@@ -0,0 +1,101 @@
+/*
+ * 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.testsuite.iud
+
+import java.text.SimpleDateFormat
+import java.util
+import java.util.concurrent.{Callable, ExecutorService, Executors, Future}
+
+import org.apache.spark.sql.test.util.QueryTest
+import org.apache.spark.sql.{DataFrame, SaveMode}
+import org.scalatest.BeforeAndAfterAll
+
+import org.apache.carbondata.core.constants.CarbonCommonConstants
+import org.apache.carbondata.core.util.CarbonProperties
+
+class TestInsertUpdateConcurrentTest extends QueryTest with BeforeAndAfterAll {
+ var df: DataFrame = _
+ private val executorService: ExecutorService = Executors.newFixedThreadPool(10)
+
+ override def beforeAll {
+ dropTable()
+ buildTestData()
+ }
+
+ override def afterAll {
+ executorService.shutdownNow()
+ dropTable()
+ }
+
+
+ private def buildTestData(): Unit = {
+
+ CarbonProperties.getInstance()
+ .addProperty(CarbonCommonConstants.CARBON_DATE_FORMAT, "yyyy-MM-dd")
+
+ // Simulate data and write to table orders
+ import sqlContext.implicits._
+
+ val sdf = new SimpleDateFormat("yyyy-MM-dd")
+ df = sqlContext.sparkSession.sparkContext.parallelize(1 to 150000)
+ .map(value => (value, new java.sql.Date(sdf.parse("2015-07-" + (value % 10 + 10)).getTime),
+ "china", "aaa" + value, "phone" + 555 * value, "ASD" + (60000 + value), 14999 + value,"ordersTable"+value))
+ .toDF("o_id", "o_date", "o_country", "o_name",
+ "o_phonetype", "o_serialname", "o_salary","o_comment")
+ createTable("orders")
+ createTable("orders_overwrite")
+ }
+
+ private def dropTable() = {
+ sql("DROP TABLE IF EXISTS orders")
+ sql("DROP TABLE IF EXISTS orders_overwrite")
+ }
+
+ private def createTable(tableName: String): Unit ={
+ df.write
+ .format("carbondata")
+ .option("tableName", tableName)
+ .option("tempCSV", "true")
+ .option("compress", "true")
+ .mode(SaveMode.Overwrite)
+ .save()
+ }
+
+ test("Concurrency test for Insert-Overwrite and update") {
+ val tasks = new java.util.ArrayList[Callable[String]]()
+ tasks.add(new QueryTask(s"insert overWrite table orders select * from orders_overwrite"))
+ tasks.add(new QueryTask("update orders set (o_country)=('newCountry') where o_country='china'"))
+ val results: util.List[Future[String]] = executorService.invokeAll(tasks)
+ assert("PASS".equals(results.get(0).get) && "FAIL".equals(results.get(1).get))
+ }
+
+ class QueryTask(query: String) extends Callable[String] {
+ override def call(): String = {
+ var result = "PASS"
+ try {
+ LOGGER.info("Executing :" + query + Thread.currentThread().getName)
+ sql(query).show()
+ } catch {
+ case _: Exception =>
+ result = "FAIL"
+ }
+ result
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/62ce5b5a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/mutation/CarbonProjectForUpdateCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/mutation/CarbonProjectForUpdateCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/mutation/CarbonProjectForUpdateCommand.scala
index bd53a66..20a6bab 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/mutation/CarbonProjectForUpdateCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/mutation/CarbonProjectForUpdateCommand.scala
@@ -28,9 +28,11 @@ import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.locks.{CarbonLockFactory, CarbonLockUtil, LockUsage}
import org.apache.carbondata.core.mutate.CarbonUpdateUtil
+import org.apache.carbondata.core.statusmanager.SegmentStatusManager
import org.apache.carbondata.core.util.CarbonProperties
import org.apache.carbondata.events.{OperationContext, OperationListenerBus, UpdateTablePostEvent, UpdateTablePreEvent}
import org.apache.carbondata.processing.loading.FailureCauses
+import org.apache.carbondata.spark.exception.ConcurrentOperationException
private[sql] case class CarbonProjectForUpdateCommand(
plan: LogicalPlan,
@@ -52,6 +54,13 @@ private[sql] case class CarbonProjectForUpdateCommand(
return Seq.empty
}
val carbonTable = CarbonEnv.getCarbonTable(databaseNameOp, tableName)(sparkSession)
+ val isLoadInProgress = SegmentStatusManager.checkIfAnyLoadInProgressForTable(carbonTable)
+ if (isLoadInProgress) {
+ val errorMessage = "Cannot run data loading and update on same table concurrently. Please " +
+ "wait for load to finish"
+ LOGGER.error(errorMessage)
+ throw new ConcurrentOperationException(errorMessage)
+ }
// trigger event for Update table
val operationContext = new OperationContext
[11/14] carbondata git commit: [CARBONDATA-2022]Fixed table alias
issue in PreAggregate
Posted by ja...@apache.org.
[CARBONDATA-2022]Fixed table alias issue in PreAggregate
**Problem:**Query with table alias is Not hitting pre Aggregate table.
Solution: Problem is table alias query is plan is coming as SubQueryAlias(alias, SubqueryAlias) ans this case is not present in tranform query plan for pre aggregate
This closes #1794
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/913837fb
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/913837fb
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/913837fb
Branch: refs/heads/carbonstore
Commit: 913837fbcf6d51f21b4f1130158ea3281b7a89b3
Parents: 36b7982
Author: BJangir <ba...@gmail.com>
Authored: Thu Jan 11 15:30:59 2018 +0530
Committer: ravipesala <ra...@gmail.com>
Committed: Fri Jan 12 19:07:29 2018 +0530
----------------------------------------------------------------------
.../TestPreAggregateTableSelection.scala | 18 ++++-
.../sql/hive/CarbonPreAggregateRules.scala | 85 ++++++++++++++++++++
2 files changed, 102 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/913837fb/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateTableSelection.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateTableSelection.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateTableSelection.scala
index 2e0dcc4..9bbba3a 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateTableSelection.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateTableSelection.scala
@@ -240,12 +240,28 @@ class TestPreAggregateTableSelection extends QueryTest with BeforeAndAfterAll {
sql("CREATE TABLE grouptable(id int, name string, city string, age string) STORED BY" +
" 'org.apache.carbondata.format' TBLPROPERTIES('dictionary_include'='name,age')")
sql(s"LOAD DATA LOCAL INPATH '$resourcesPath/measureinsertintotest.csv' into table grouptable")
- sql("create datamap agg9 on table grouptable using 'preaggregate' as select sum(id) from grouptable group by city")
+ sql(
+ "create datamap agg9 on table grouptable using 'preaggregate' as select sum(id) from grouptable group by city")
val df = sql("select sum(id) from grouptable group by city")
preAggTableValidator(df.queryExecution.analyzed, "grouptable_agg9")
checkAnswer(df, Seq(Row(3), Row(3), Row(4), Row(7)))
}
+ test("test PreAggregate table selection 30") {
+ val df = sql("select a.name from mainTable a group by a.name")
+ preAggTableValidator(df.queryExecution.analyzed, "maintable_agg0")
+ }
+
+ test("test PreAggregate table selection 31") {
+ val df = sql("select a.name as newName from mainTable a group by a.name")
+ preAggTableValidator(df.queryExecution.analyzed, "maintable_agg0")
+ }
+
+ test("test PreAggregate table selection 32") {
+ val df = sql("select a.name as newName from mainTable a where a.name='vishal' group by a.name")
+ preAggTableValidator(df.queryExecution.analyzed, "maintable_agg0")
+ }
+
override def afterAll: Unit = {
sql("drop table if exists mainTable")
sql("drop table if exists lineitem")
http://git-wip-us.apache.org/repos/asf/carbondata/blob/913837fb/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonPreAggregateRules.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonPreAggregateRules.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonPreAggregateRules.scala
index 8811a4e..de554c5 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonPreAggregateRules.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonPreAggregateRules.scala
@@ -260,6 +260,48 @@ case class CarbonPreAggregateQueryRules(sparkSession: SparkSession) extends Rule
*/
def transformPreAggQueryPlan(logicalPlan: LogicalPlan): LogicalPlan = {
val updatedPlan = logicalPlan.transform {
+ case agg@Aggregate(
+ grExp,
+ aggExp,
+ CarbonSubqueryAlias(_, child@CarbonSubqueryAlias(_, l: LogicalRelation)))
+ if l.relation.isInstanceOf[CarbonDatasourceHadoopRelation] &&
+ l.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonRelation.
+ metaData.hasAggregateDataMapSchema =>
+ val carbonTable = getCarbonTable(l)
+ val list = scala.collection.mutable.HashSet.empty[QueryColumn]
+ val aggregateExpressions = scala.collection.mutable.HashSet.empty[AggregateExpression]
+ val isValidPlan = extractQueryColumnsFromAggExpression(
+ grExp,
+ aggExp,
+ carbonTable,
+ list,
+ aggregateExpressions)
+ if(isValidPlan) {
+ val (aggDataMapSchema, childPlan) = getChildDataMapForTransformation(list,
+ aggregateExpressions,
+ carbonTable,
+ agg)
+ if(null != aggDataMapSchema && null!= childPlan) {
+ val attributes = childPlan.output.asInstanceOf[Seq[AttributeReference]]
+ val (updatedGroupExp, updatedAggExp, newChild, None) =
+ getUpdatedExpressions(grExp,
+ aggExp,
+ child,
+ None,
+ aggDataMapSchema,
+ attributes,
+ childPlan,
+ carbonTable,
+ agg)
+ Aggregate(updatedGroupExp,
+ updatedAggExp,
+ newChild)
+ } else {
+ agg
+ }
+ } else {
+ agg
+ }
// case for aggregation query
case agg@Aggregate(
grExp,
@@ -352,6 +394,49 @@ case class CarbonPreAggregateQueryRules(sparkSession: SparkSession) extends Rule
} else {
agg
}
+ case agg@Aggregate(
+ grExp,
+ aggExp,
+ Filter(expression, CarbonSubqueryAlias(_, child@CarbonSubqueryAlias(_, l: LogicalRelation))))
+ if l.relation.isInstanceOf[CarbonDatasourceHadoopRelation] &&
+ l.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonRelation.
+ metaData.hasAggregateDataMapSchema =>
+ val carbonTable = getCarbonTable(l)
+ val list = scala.collection.mutable.HashSet.empty[QueryColumn]
+ val aggregateExpressions = scala.collection.mutable.HashSet.empty[AggregateExpression]
+ val isValidPlan = extractQueryColumnsFromAggExpression(
+ grExp,
+ aggExp,
+ carbonTable,
+ list,
+ aggregateExpressions)
+ if(isValidPlan) {
+ val (aggDataMapSchema, childPlan) = getChildDataMapForTransformation(list,
+ aggregateExpressions,
+ carbonTable,
+ agg)
+ if(null != aggDataMapSchema && null!= childPlan) {
+ val attributes = childPlan.output.asInstanceOf[Seq[AttributeReference]]
+ val (updatedGroupExp, updatedAggExp, newChild, None) =
+ getUpdatedExpressions(grExp,
+ aggExp,
+ child,
+ None,
+ aggDataMapSchema,
+ attributes,
+ childPlan,
+ carbonTable,
+ agg)
+ Aggregate(updatedGroupExp,
+ updatedAggExp,
+ newChild)
+ } else {
+ agg
+ }
+ } else {
+ agg
+ }
+
}
updatedPlan
}
[14/14] carbondata git commit: [CARBONDATA-1965]removed sort_scope
from dynamic configuration in carbondata using set-reset as it is not
configured by set
Posted by ja...@apache.org.
[CARBONDATA-1965]removed sort_scope from dynamic configuration in carbondata using set-reset as it is not configured by set
This closes #1784
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/b8bcb0e0
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/b8bcb0e0
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/b8bcb0e0
Branch: refs/heads/carbonstore
Commit: b8bcb0e0eec85f31b7707600da7057a830d46158
Parents: df27831
Author: vandana <va...@gmail.com>
Authored: Wed Jan 10 15:38:40 2018 +0530
Committer: Jacky Li <ja...@qq.com>
Committed: Sat Jan 13 16:02:38 2018 +0800
----------------------------------------------------------------------
docs/configuration-parameters.md | 2 --
1 file changed, 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/b8bcb0e0/docs/configuration-parameters.md
----------------------------------------------------------------------
diff --git a/docs/configuration-parameters.md b/docs/configuration-parameters.md
index 05aede7..338854f 100644
--- a/docs/configuration-parameters.md
+++ b/docs/configuration-parameters.md
@@ -191,11 +191,9 @@ RESET
| carbon.options.bad.records.logger.enable | To enable or disable bad record logger. |
| carbon.options.bad.records.action | This property can have four types of actions for bad records FORCE, REDIRECT, IGNORE and FAIL. If set to FORCE then it auto-corrects the data by storing the bad records as NULL. If set to REDIRECT then bad records are written to the raw CSV instead of being loaded. If set to IGNORE then bad records are neither loaded nor written to the raw CSV. If set to FAIL then data loading fails if any bad records are found. |
| carbon.options.is.empty.data.bad.record | If false, then empty ("" or '' or ,,) data will not be considered as bad record and vice versa. |
-| carbon.options.sort.scope | This property can have four possible values BATCH_SORT, LOCAL_SORT, GLOBAL_SORT and NO_SORT. If set to BATCH_SORT, the sorting scope is smaller and more index tree will be created,thus loading is faster but query maybe slower. If set to LOCAL_SORT, the sorting scope is bigger and one index tree per data node will be created, thus loading is slower but query is faster. If set to GLOBAL_SORT, the sorting scope is bigger and one index tree per task will be created, thus loading is slower but query is faster. If set to NO_SORT data will be loaded in unsorted manner. |
| carbon.options.batch.sort.size.inmb | Size of batch data to keep in memory, as a thumb rule it supposed to be less than 45% of sort.inmemory.size.inmb otherwise it may spill intermediate data to disk. |
| carbon.options.single.pass | Single Pass Loading enables single job to finish data loading with dictionary generation on the fly. It enhances performance in the scenarios where the subsequent data loading after initial load involves fewer incremental updates on the dictionary. This option specifies whether to use single pass for loading data or not. By default this option is set to FALSE. |
| carbon.options.bad.record.path | Specifies the HDFS path where bad records needs to be stored. |
-| carbon.options.global.sort.partitions | The Number of partitions to use when shuffling data for sort. If user don't configurate or configurate it less than 1, it uses the number of map tasks as reduce tasks. In general, we recommend 2-3 tasks per CPU core in your cluster. |
| carbon.custom.block.distribution | Specifies whether to use the Spark or Carbon block distribution feature. |
| enable.unsafe.sort | Specifies whether to use unsafe sort during data loading. Unsafe sort reduces the garbage collection during data load operation, resulting in better performance. |
[05/14] carbondata git commit: [CARBONDATA-2011] Fix
ClassCastException in CarbonStreamingQueryListener
Posted by ja...@apache.org.
[CARBONDATA-2011] Fix ClassCastException in CarbonStreamingQueryListener
Fix ClassCastException in CarbonStreamingQueryListener for Spark 2.1 integration
This closes #1779
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/3eb1e2b3
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/3eb1e2b3
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/3eb1e2b3
Branch: refs/heads/carbonstore
Commit: 3eb1e2b30d64137293dd1d7985883c0ef75646ae
Parents: 082a355
Author: QiangCai <qi...@qq.com>
Authored: Tue Jan 9 19:07:19 2018 +0800
Committer: Jacky Li <ja...@qq.com>
Committed: Thu Jan 11 09:28:42 2018 +0800
----------------------------------------------------------------------
.../sql/execution/streaming/CarbonStreamingQueryListener.scala | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/3eb1e2b3/streaming/src/main/scala/org/apache/spark/sql/execution/streaming/CarbonStreamingQueryListener.scala
----------------------------------------------------------------------
diff --git a/streaming/src/main/scala/org/apache/spark/sql/execution/streaming/CarbonStreamingQueryListener.scala b/streaming/src/main/scala/org/apache/spark/sql/execution/streaming/CarbonStreamingQueryListener.scala
index 07ef8ca..2f911c5 100644
--- a/streaming/src/main/scala/org/apache/spark/sql/execution/streaming/CarbonStreamingQueryListener.scala
+++ b/streaming/src/main/scala/org/apache/spark/sql/execution/streaming/CarbonStreamingQueryListener.scala
@@ -35,7 +35,7 @@ class CarbonStreamingQueryListener(spark: SparkSession) extends StreamingQueryLi
override def onQueryStarted(event: StreamingQueryListener.QueryStartedEvent): Unit = {
val streamQuery = spark.streams.get(event.id)
- val qry = if (SPARK_VERSION.startsWith("2.1")) {
+ val qry = if (streamQuery.isInstanceOf[StreamExecution]) {
// adapt spark 2.1
streamQuery.asInstanceOf[StreamExecution]
} else {
[12/14] carbondata git commit: [CARBONDATA-2017] Fix input path
checking when loading data from multiple paths
Posted by ja...@apache.org.
[CARBONDATA-2017] Fix input path checking when loading data from multiple paths
This closes #1787
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/6d77fdff
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/6d77fdff
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/6d77fdff
Branch: refs/heads/carbonstore
Commit: 6d77fdff444952a9c9ac8b02c529f9e387164be2
Parents: 913837f
Author: Manhua <ke...@qq.com>
Authored: Wed Jan 10 21:22:57 2018 +0800
Committer: Jacky Li <ja...@qq.com>
Committed: Fri Jan 12 22:24:08 2018 +0800
----------------------------------------------------------------------
.../src/main/scala/org/apache/spark/util/FileUtils.scala | 3 ++-
.../sql/execution/command/management/CarbonLoadDataCommand.scala | 3 +--
2 files changed, 3 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/6d77fdff/integration/spark-common/src/main/scala/org/apache/spark/util/FileUtils.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/spark/util/FileUtils.scala b/integration/spark-common/src/main/scala/org/apache/spark/util/FileUtils.scala
index b60d030..95ba318 100644
--- a/integration/spark-common/src/main/scala/org/apache/spark/util/FileUtils.scala
+++ b/integration/spark-common/src/main/scala/org/apache/spark/util/FileUtils.scala
@@ -73,7 +73,8 @@ object FileUtils {
val stringBuild = new StringBuilder()
val filePaths = inputPath.split(",")
for (i <- 0 until filePaths.size) {
- val fileType = FileFactory.getFileType(filePaths(i))
+ val filePath = CarbonUtil.checkAndAppendHDFSUrl(filePaths(i))
+ val fileType = FileFactory.getFileType(filePath)
val carbonFile = FileFactory.getCarbonFile(filePaths(i), fileType, hadoopConf)
if (!carbonFile.exists()) {
throw new DataLoadingException(
http://git-wip-us.apache.org/repos/asf/carbondata/blob/6d77fdff/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala
index 1828557..9577615 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala
@@ -147,8 +147,7 @@ case class CarbonLoadDataCommand(
val factPath = if (dataFrame.isDefined) {
""
} else {
- FileUtils.getPaths(
- CarbonUtil.checkAndAppendHDFSUrl(factPathFromUser), hadoopConf)
+ FileUtils.getPaths(factPathFromUser, hadoopConf)
}
carbonLoadModel.setFactFilePath(factPath)
carbonLoadModel.setAggLoadRequest(
[04/14] carbondata git commit: [CARBONDATA-1703] Fix null handling in
"IN" and "NOT IN" expressions
Posted by ja...@apache.org.
[CARBONDATA-1703] Fix null handling in "IN" and "NOT IN" expressions
This closes #1650
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/082a3559
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/082a3559
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/082a3559
Branch: refs/heads/carbonstore
Commit: 082a3559aa4d36b1263234e290ea77e5eead8db2
Parents: 62ce5b5
Author: Geetika Gupta <ge...@knoldus.in>
Authored: Wed Dec 13 12:03:22 2017 +0530
Committer: Venkata Ramana G <ra...@huawei.com>
Committed: Tue Jan 9 19:48:03 2018 +0530
----------------------------------------------------------------------
.../ExpressionWithNullTestCase.scala | 21 +++++++++++++++-----
.../spark/sql/optimizer/CarbonFilters.scala | 21 ++++++++++++++------
.../BooleanDataTypesFilterTest.scala | 4 ++--
3 files changed, 33 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/082a3559/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/detailquery/ExpressionWithNullTestCase.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/detailquery/ExpressionWithNullTestCase.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/detailquery/ExpressionWithNullTestCase.scala
index 2024018..a2d124d 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/detailquery/ExpressionWithNullTestCase.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/detailquery/ExpressionWithNullTestCase.scala
@@ -48,9 +48,9 @@ class ExpressionWithNullTestCase extends QueryTest with BeforeAndAfterAll {
checkAnswer(sql("select * from expression_test where id in (1,2,'', NULL, ' ')"), sql("select * from expression_test_hive where id in (1,2,' ', NULL, ' ')"))
checkAnswer(sql("select * from expression_test where id in (1,2,'')"), sql("select * from expression_test_hive where id in (1,2,'')"))
checkAnswer(sql("select * from expression_test where id in ('')"), sql("select * from expression_test_hive where id in ('')"))
-// checkAnswer(sql("select * from expression_test where number in (null)"), sql("select * from expression_test_hive where number in (null)"))
+ checkAnswer(sql("select * from expression_test where number in (NULL)"), sql("select * from expression_test_hive where number in (null)"))
checkAnswer(sql("select * from expression_test where number in (2)"), sql("select * from expression_test_hive where number in (2)"))
-// checkAnswer(sql("select * from expression_test where number in (1,null)"), sql("select * from expression_test_hive where number in (1,null)"))
+ checkAnswer(sql("select * from expression_test where number in (1,null)"), sql("select * from expression_test_hive where number in (1,null)"))
checkAnswer(sql("select * from expression where number in (1,null)"), sql("select * from expression_hive where number in (1,null)"))
checkAnswer(sql("select * from expression where id in (3)"), sql("select * from expression_hive where id in (3)"))
checkAnswer(sql("select * from expression where id in ('2')"), sql("select * from expression_hive where id in ('2')"))
@@ -59,14 +59,20 @@ class ExpressionWithNullTestCase extends QueryTest with BeforeAndAfterAll {
checkAnswer(sql("select * from expression_test where id in ('2')"), sql("select * from expression_test_hive where id in ('2')"))
checkAnswer(sql("select * from expression_test where id in (cast('2' as int))"), sql("select * from expression_test_hive where id in (cast('2' as int))"))
checkAnswer(sql("select * from expression_test where id in (cast('null' as int))"), sql("select * from expression_test_hive where id in (cast('null' as int))"))
+ checkAnswer(sql("select * from expression_test where id in (1,2,NULL)"), sql("select * from expression_test_hive where id in (1,2,NULL)"))
+ checkAnswer(sql("select * from expression_test where id in (NULL)"), sql("select * from expression_test_hive where id in (NULL)"))
+ checkAnswer(sql("select * from expression_test where number in (cast('null' as int))"), sql("select * from expression_test_hive where number in (cast('null' as int))"))
+ checkAnswer(sql("select * from expression_test where number in (1,2, cast('NULL' as int), cast('3' as int))"), sql("select * from expression_test_hive where number in (1,2, cast('NULL' as int),cast('3' as int))"))
+ checkAnswer(sql("select * from expression_test where cast(number as int) IN(1,null)"), sql("select * from expression_test_hive where cast(number as int) IN(1,null)"))
+
}
test("test to check not in expression with null values") {
checkAnswer(sql("select * from expression_test where id not in (1,2,'', NULL, ' ')"), sql("select * from expression_test_hive where id not in (1,2,' ', NULL, ' ')"))
checkAnswer(sql("select * from expression_test where id not in (1,2,'')"), sql("select * from expression_test_hive where id not in (1,2,'')"))
checkAnswer(sql("select * from expression_test where id not in ('')"), sql("select * from expression_test_hive where id not in ('')"))
-// checkAnswer(sql("select * from expression_test where number not in (null)"), sql("select * from expression_test_hive where number not in (null)"))
-// checkAnswer(sql("select * from expression_test where number not in (1,null)"), sql("select * from expression_test_hive where number not in (1,null)"))
+ checkAnswer(sql("select * from expression_test where number not in (null)"), sql("select * from expression_test_hive where number not in (null)"))
+ checkAnswer(sql("select * from expression_test where number not in (1,null)"), sql("select * from expression_test_hive where number not in (1,null)"))
checkAnswer(sql("select * from expression where number not in (1,null)"), sql("select * from expression_hive where number not in (1,null)"))
checkAnswer(sql("select * from expression where id not in (3)"), sql("select * from expression_hive where id not in (3)"))
checkAnswer(sql("select * from expression where id not in ('2')"), sql("select * from expression_hive where id not in ('2')"))
@@ -74,7 +80,12 @@ class ExpressionWithNullTestCase extends QueryTest with BeforeAndAfterAll {
checkAnswer(sql("select * from expression_test where id not in (3)"), sql("select * from expression_test_hive where id not in (3)"))
checkAnswer(sql("select * from expression_test where id not in ('2')"), sql("select * from expression_test_hive where id not in ('2')"))
checkAnswer(sql("select * from expression_test where id not in (cast('2' as int))"), sql("select * from expression_test_hive where id not in (cast('2' as int))"))
-// checkAnswer(sql("select * from expression_test where id not in (cast('null' as int))"), sql("select * from expression_test_hive where id not in (cast('null' as int))"))
+ checkAnswer(sql("select * from expression_test where id not in (cast('null' as int))"), sql("select * from expression_test_hive where id not in (cast('null' as int))"))
+ checkAnswer(sql("select * from expression_test where id not in (1,2,NULL)"), sql("select * from expression_test_hive where id not in (1,2,NULL)"))
+ checkAnswer(sql("select * from expression_test where id not in (NULL)"), sql("select * from expression_test_hive where id not in (NULL)"))
+ checkAnswer(sql("select * from expression_test where number not in (2, null)"), sql("select * from expression_test_hive where number not in (2, null)"))
+ checkAnswer(sql("select * from expression_test where number not in (cast('2' as int), cast('null' as int))"), sql("select * from expression_test_hive where number not in (cast('2' as int), cast('null' as int))"))
+
}
test("test to check equals expression with null values") {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/082a3559/integration/spark2/src/main/scala/org/apache/spark/sql/optimizer/CarbonFilters.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/optimizer/CarbonFilters.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/optimizer/CarbonFilters.scala
index 5027a66..341b513 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/optimizer/CarbonFilters.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/optimizer/CarbonFilters.scala
@@ -80,13 +80,22 @@ object CarbonFilters {
Some(new LessThanEqualToExpression(getCarbonExpression(name),
getCarbonLiteralExpression(name, value)))
case sources.In(name, values) =>
- Some(new InExpression(getCarbonExpression(name),
- new ListExpression(
- convertToJavaList(values.map(f => getCarbonLiteralExpression(name, f)).toList))))
+ if (values.length == 1 && values(0) == null) {
+ Some(new FalseExpression(getCarbonExpression(name)))
+ } else {
+ Some(new InExpression(getCarbonExpression(name),
+ new ListExpression(
+ convertToJavaList(values.filterNot(_ == null)
+ .map(filterValues => getCarbonLiteralExpression(name, filterValues)).toList))))
+ }
case sources.Not(sources.In(name, values)) =>
- Some(new NotInExpression(getCarbonExpression(name),
- new ListExpression(
- convertToJavaList(values.map(f => getCarbonLiteralExpression(name, f)).toList))))
+ if (values.contains(null)) {
+ Some(new FalseExpression(getCarbonExpression(name)))
+ } else {
+ Some(new NotInExpression(getCarbonExpression(name),
+ new ListExpression(
+ convertToJavaList(values.map(f => getCarbonLiteralExpression(name, f)).toList))))
+ }
case sources.IsNull(name) =>
Some(new EqualToExpression(getCarbonExpression(name),
getCarbonLiteralExpression(name, null), true))
http://git-wip-us.apache.org/repos/asf/carbondata/blob/082a3559/integration/spark2/src/test/scala/org/apache/carbondata/spark/testsuite/booleantype/BooleanDataTypesFilterTest.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/test/scala/org/apache/carbondata/spark/testsuite/booleantype/BooleanDataTypesFilterTest.scala b/integration/spark2/src/test/scala/org/apache/carbondata/spark/testsuite/booleantype/BooleanDataTypesFilterTest.scala
index 6149ac9..cdcf018 100644
--- a/integration/spark2/src/test/scala/org/apache/carbondata/spark/testsuite/booleantype/BooleanDataTypesFilterTest.scala
+++ b/integration/spark2/src/test/scala/org/apache/carbondata/spark/testsuite/booleantype/BooleanDataTypesFilterTest.scala
@@ -159,10 +159,10 @@ class BooleanDataTypesFilterTest extends QueryTest with BeforeAndAfterEach with
Row(4))
checkAnswer(sql("select * from carbon_table where booleanField not in (null)"),
- Seq(Row(true), Row(true), Row(true), Row(true), Row(false), Row(false), Row(false), Row(false)))
+ Seq.empty)
checkAnswer(sql("select count(*) from carbon_table where booleanField not in (null)"),
- Row(8))
+ Row(0))
checkAnswer(sql("select count(*) from carbon_table where booleanField not in (true,false)"),
Row(0))
[02/14] carbondata git commit: [CARBONDATA-1954] [Pre-Aggregate]
CarbonHiveMetastore updated while dropping the Pre-Aggregate table & code
refactored
Posted by ja...@apache.org.
[CARBONDATA-1954] [Pre-Aggregate] CarbonHiveMetastore updated while dropping the Pre-Aggregate table & code refactored
1. To update CarbonHiveMetastore similar function was already there . Removed duplicate function defination and updated the caller.
2. code refactored so that during droping a pre-aggregate table only metadata will be deleted if processMetadata() is called.
This closes #1743
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/d33d3473
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/d33d3473
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/d33d3473
Branch: refs/heads/carbonstore
Commit: d33d3473e53fed913a9794e150050f2007acde63
Parents: 1b72a02
Author: rahulforallp <ra...@knoldus.in>
Authored: Fri Dec 29 20:01:45 2017 +0530
Committer: ravipesala <ra...@gmail.com>
Committed: Mon Jan 8 22:28:29 2018 +0530
----------------------------------------------------------------------
.../testsuite/datamap/TestDataMapCommand.scala | 54 +++++++++++++++
.../org/apache/spark/sql/CarbonSession.scala | 3 -
.../datamap/CarbonDropDataMapCommand.scala | 19 +++++-
.../command/datamap/DataMapListeners.scala | 72 --------------------
.../command/table/CarbonDropTableCommand.scala | 38 ++++++++++-
.../spark/sql/hive/CarbonHiveMetaStore.scala | 23 +------
6 files changed, 109 insertions(+), 100 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d33d3473/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/TestDataMapCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/TestDataMapCommand.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/TestDataMapCommand.scala
index d61971e..c38e6cf 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/TestDataMapCommand.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/TestDataMapCommand.scala
@@ -20,7 +20,9 @@ package org.apache.carbondata.spark.testsuite.datamap
import org.apache.spark.sql.test.util.QueryTest
import org.scalatest.BeforeAndAfterAll
+import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.metadata.CarbonMetadata
+import org.apache.carbondata.core.util.CarbonProperties
class TestDataMapCommand extends QueryTest with BeforeAndAfterAll {
@@ -76,6 +78,58 @@ class TestDataMapCommand extends QueryTest with BeforeAndAfterAll {
assert(dataMapSchemaList.get(2).getChildSchema.getTableName.equals("datamaptest_datamap3"))
}
+ test("check hivemetastore after drop datamap") {
+ try {
+ CarbonProperties.getInstance()
+ .addProperty(CarbonCommonConstants.ENABLE_HIVE_SCHEMA_META_STORE,
+ "true")
+ sql("drop datamap if exists datamap_hiveMetaStoreTable on table hiveMetaStoreTable")
+ sql("drop table if exists hiveMetaStoreTable")
+ sql("create table hiveMetaStoreTable (a string, b string, c string) stored by 'carbondata'")
+
+ sql(
+ "create datamap datamap_hiveMetaStoreTable on table hiveMetaStoreTable using 'preaggregate' dmproperties('key'='value') as select count(a) from hiveMetaStoreTable")
+ checkExistence(sql("show datamap on table hiveMetaStoreTable"), true, "datamap_hiveMetaStoreTable")
+
+ sql("drop datamap datamap_hiveMetaStoreTable on table hiveMetaStoreTable")
+ checkExistence(sql("show datamap on table hiveMetaStoreTable"), false, "datamap_hiveMetaStoreTable")
+
+ }
+ finally {
+ sql("drop table hiveMetaStoreTable")
+ CarbonProperties.getInstance()
+ .addProperty(CarbonCommonConstants.ENABLE_HIVE_SCHEMA_META_STORE,
+ CarbonCommonConstants.ENABLE_HIVE_SCHEMA_META_STORE_DEFAULT)
+ }
+ }
+
+ test("drop the table having pre-aggregate"){
+ try {
+ CarbonProperties.getInstance()
+ .addProperty(CarbonCommonConstants.ENABLE_HIVE_SCHEMA_META_STORE,
+ "true")
+ sql("drop datamap if exists datamap_hiveMetaStoreTable_1 on table hiveMetaStoreTable_1")
+ sql("drop table if exists hiveMetaStoreTable_1")
+ sql("create table hiveMetaStoreTable_1 (a string, b string, c string) stored by 'carbondata'")
+
+ sql(
+ "create datamap datamap_hiveMetaStoreTable_1 on table hiveMetaStoreTable_1 using 'preaggregate' dmproperties('key'='value') as select count(a) from hiveMetaStoreTable_1")
+
+ checkExistence(sql("show datamap on table hiveMetaStoreTable_1"),
+ true,
+ "datamap_hiveMetaStoreTable_1")
+
+ sql("drop table hiveMetaStoreTable_1")
+
+ checkExistence(sql("show tables"), false, "datamap_hiveMetaStoreTable_1")
+ }
+ finally {
+ CarbonProperties.getInstance()
+ .addProperty(CarbonCommonConstants.ENABLE_HIVE_SCHEMA_META_STORE,
+ CarbonCommonConstants.ENABLE_HIVE_SCHEMA_META_STORE_DEFAULT)
+ }
+ }
+
test("test datamap create with preagg with duplicate name") {
intercept[Exception] {
sql(
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d33d3473/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSession.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSession.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSession.scala
index 7ee3434..34e37c5 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSession.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSession.scala
@@ -24,7 +24,6 @@ import org.apache.hadoop.conf.Configuration
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.scheduler.{SparkListener, SparkListenerApplicationEnd}
import org.apache.spark.sql.SparkSession.Builder
-import org.apache.spark.sql.execution.command.datamap.{DataMapDropTablePostListener, DropDataMapPostListener}
import org.apache.spark.sql.execution.command.preaaggregate._
import org.apache.spark.sql.execution.streaming.CarbonStreamingQueryListener
import org.apache.spark.sql.hive.execution.command.CarbonSetCommand
@@ -250,7 +249,6 @@ object CarbonSession {
def initListeners(): Unit = {
OperationListenerBus.getInstance()
- .addListener(classOf[DropTablePostEvent], DataMapDropTablePostListener)
.addListener(classOf[LoadTablePreStatusUpdateEvent], LoadPostAggregateListener)
.addListener(classOf[DeleteSegmentByIdPreEvent], PreAggregateDeleteSegmentByIdPreListener)
.addListener(classOf[DeleteSegmentByDatePreEvent], PreAggregateDeleteSegmentByDatePreListener)
@@ -261,7 +259,6 @@ object CarbonSession {
.addListener(classOf[AlterTableRenamePreEvent], PreAggregateRenameTablePreListener)
.addListener(classOf[AlterTableDataTypeChangePreEvent], PreAggregateDataTypeChangePreListener)
.addListener(classOf[AlterTableAddColumnPreEvent], PreAggregateAddColumnsPreListener)
- .addListener(classOf[DropDataMapPostEvent], DropDataMapPostListener)
.addListener(classOf[LoadTablePreExecutionEvent], LoadPreAggregateTablePreListener)
.addListener(classOf[AlterTableCompactionPreStatusUpdateEvent],
AlterPreAggregateTableCompactionPostListener)
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d33d3473/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDropDataMapCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDropDataMapCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDropDataMapCommand.scala
index 7f68b05..59aa322 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDropDataMapCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDropDataMapCommand.scala
@@ -25,16 +25,15 @@ import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.analysis.NoSuchTableException
import org.apache.spark.sql.execution.command.AtomicRunnableCommand
import org.apache.spark.sql.execution.command.preaaggregate.PreAggregateUtil
+import org.apache.spark.sql.execution.command.table.CarbonDropTableCommand
import org.apache.spark.sql.hive.CarbonRelation
import org.apache.carbondata.common.logging.{LogService, LogServiceFactory}
-import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.datamap.DataMapStoreManager
import org.apache.carbondata.core.locks.{CarbonLockUtil, ICarbonLock, LockUsage}
import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier
import org.apache.carbondata.core.metadata.converter.ThriftWrapperSchemaConverterImpl
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
-import org.apache.carbondata.core.util.CarbonProperties
import org.apache.carbondata.events._
@@ -103,6 +102,16 @@ case class CarbonDropDataMapCommand(
carbonTable.get.getTableInfo,
dbName,
tableName))(sparkSession)
+ if (dataMapSchema.isDefined) {
+ if (dataMapSchema.get._1.getRelationIdentifier != null) {
+ CarbonDropTableCommand(
+ ifExistsSet = true,
+ Some(dataMapSchema.get._1.getRelationIdentifier.getDatabaseName),
+ dataMapSchema.get._1.getRelationIdentifier.getTableName,
+ dropChildTable = true
+ ).processMetadata(sparkSession)
+ }
+ }
// fires the event after dropping datamap from main table schema
val dropDataMapPostEvent =
DropDataMapPostEvent(
@@ -136,6 +145,12 @@ case class CarbonDropDataMapCommand(
// delete the table folder
val tableIdentifier = CarbonEnv.getIdentifier(databaseNameOp, tableName)(sparkSession)
DataMapStoreManager.getInstance().clearDataMap(tableIdentifier, dataMapName)
+ CarbonDropTableCommand(
+ ifExistsSet = true,
+ databaseNameOp,
+ dataMapName,
+ dropChildTable = true
+ ).processData(sparkSession)
Seq.empty
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d33d3473/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/DataMapListeners.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/DataMapListeners.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/DataMapListeners.scala
deleted file mode 100644
index d37ca0a..0000000
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/DataMapListeners.scala
+++ /dev/null
@@ -1,72 +0,0 @@
-/*
- * 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.execution.command.datamap
-
-import scala.collection.JavaConverters._
-
-import org.apache.spark.sql.execution.command.AtomicRunnableCommand
-import org.apache.spark.sql.execution.command.table.CarbonDropTableCommand
-
-import org.apache.carbondata.events.{DropDataMapPostEvent, DropTablePostEvent, Event, OperationContext, OperationEventListener}
-
-object DataMapDropTablePostListener extends OperationEventListener {
-
- /**
- * Called on DropTablePostEvent occurrence
- */
- override def onEvent(event: Event, operationContext: OperationContext): Unit = {
- val dropPostEvent = event.asInstanceOf[DropTablePostEvent]
- val carbonTable = dropPostEvent.carbonTable
- val sparkSession = dropPostEvent.sparkSession
- if (carbonTable.hasDataMapSchema) {
- // drop all child tables
- val childSchemas = carbonTable.getTableInfo.getDataMapSchemaList
- childSchemas.asScala
- .filter(_.getRelationIdentifier != null)
- .foreach { childSchema =>
- CarbonDropTableCommand(
- ifExistsSet = true,
- Some(childSchema.getRelationIdentifier.getDatabaseName),
- childSchema.getRelationIdentifier.getTableName,
- dropChildTable = true
- ).run(sparkSession)
- }
- }
- }
-}
-
-object DropDataMapPostListener extends OperationEventListener {
-
- /**
- * Called on DropDataMapPostEvent occurrence
- */
- override def onEvent(event: Event, operationContext: OperationContext): Unit = {
- val dropPostEvent = event.asInstanceOf[DropDataMapPostEvent]
- val dataMapSchema = dropPostEvent.dataMapSchema
- val sparkSession = dropPostEvent.sparkSession
- if (dataMapSchema.isDefined) {
- if (dataMapSchema.get.getRelationIdentifier != null) {
- CarbonDropTableCommand(
- ifExistsSet = true,
- Some(dataMapSchema.get.getRelationIdentifier.getDatabaseName),
- dataMapSchema.get.getRelationIdentifier.getTableName,
- dropChildTable = true
- ).run(sparkSession)
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d33d3473/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonDropTableCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonDropTableCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonDropTableCommand.scala
index aaad207..312e8b0 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonDropTableCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonDropTableCommand.scala
@@ -17,9 +17,11 @@
package org.apache.spark.sql.execution.command.table
+import scala.collection.JavaConverters._
import scala.collection.mutable.ListBuffer
import org.apache.spark.sql.{AnalysisException, CarbonEnv, Row, SparkSession}
+import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.analysis.NoSuchTableException
import org.apache.spark.sql.execution.command.AtomicRunnableCommand
import org.apache.spark.sql.util.CarbonException
@@ -32,7 +34,6 @@ import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.statusmanager.SegmentStatusManager
import org.apache.carbondata.core.util.CarbonUtil
import org.apache.carbondata.events._
-import org.apache.carbondata.spark.util.CommonUtil
case class CarbonDropTableCommand(
ifExistsSet: Boolean,
@@ -42,6 +43,7 @@ case class CarbonDropTableCommand(
extends AtomicRunnableCommand {
var carbonTable: CarbonTable = _
+ var childTables : Seq[CarbonTable] = Seq.empty
override def processMetadata(sparkSession: SparkSession): Seq[Row] = {
val LOGGER: LogService = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
@@ -83,6 +85,27 @@ case class CarbonDropTableCommand(
}
CarbonEnv.getInstance(sparkSession).carbonMetastore.dropTable(identifier)(sparkSession)
+ if (carbonTable.hasDataMapSchema) {
+ // drop all child tables
+ val childSchemas = carbonTable.getTableInfo.getDataMapSchemaList
+
+ childTables = childSchemas.asScala
+ .filter(_.getRelationIdentifier != null)
+ .map { childSchema =>
+ val childTable =
+ CarbonEnv.getCarbonTable(
+ TableIdentifier(childSchema.getRelationIdentifier.getTableName,
+ Some(childSchema.getRelationIdentifier.getDatabaseName)))(sparkSession)
+ CarbonDropTableCommand(
+ ifExistsSet = true,
+ Some(childSchema.getRelationIdentifier.getDatabaseName),
+ childSchema.getRelationIdentifier.getTableName,
+ dropChildTable = true
+ ).processMetadata(sparkSession)
+ childTable
+ }
+ }
+
// fires the event after dropping main table
val dropTablePostEvent: DropTablePostEvent =
DropTablePostEvent(
@@ -123,6 +146,19 @@ case class CarbonDropTableCommand(
val file = FileFactory.getCarbonFile(tablePath, fileType)
CarbonUtil.deleteFoldersAndFilesSilent(file)
}
+ if (carbonTable.hasDataMapSchema && childTables.nonEmpty) {
+ // drop all child tables
+ childTables.foreach { childTable =>
+ val carbonDropCommand = CarbonDropTableCommand(
+ ifExistsSet = true,
+ Some(childTable.getDatabaseName),
+ childTable.getTableName,
+ dropChildTable = true
+ )
+ carbonDropCommand.carbonTable = childTable
+ carbonDropCommand.processData(sparkSession)
+ }
+ }
}
Seq.empty
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d33d3473/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonHiveMetaStore.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonHiveMetaStore.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonHiveMetaStore.scala
index d5ac5ae..a2d1064 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonHiveMetaStore.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonHiveMetaStore.scala
@@ -146,7 +146,7 @@ class CarbonHiveMetaStore extends CarbonFileMetastore {
thriftTableInfo: org.apache.carbondata.format.TableInfo,
carbonTablePath: String)(sparkSession: SparkSession): String = {
val schemaConverter = new ThriftWrapperSchemaConverterImpl
- updateHiveMetaStoreForDataMap(newTableIdentifier,
+ updateHiveMetaStoreForAlter(newTableIdentifier,
oldTableIdentifier,
thriftTableInfo,
carbonTablePath,
@@ -180,27 +180,6 @@ class CarbonHiveMetaStore extends CarbonFileMetastore {
CarbonStorePath.getCarbonTablePath(oldTablePath, newTableIdentifier).getPath
}
- private def updateHiveMetaStoreForDataMap(newTableIdentifier: CarbonTableIdentifier,
- oldTableIdentifier: CarbonTableIdentifier,
- thriftTableInfo: format.TableInfo,
- tablePath: String,
- sparkSession: SparkSession,
- schemaConverter: ThriftWrapperSchemaConverterImpl) = {
- val newTablePath =
- CarbonUtil.getNewTablePath(new Path(tablePath), newTableIdentifier.getTableName)
- val wrapperTableInfo = schemaConverter.fromExternalToWrapperTableInfo(
- thriftTableInfo,
- newTableIdentifier.getDatabaseName,
- newTableIdentifier.getTableName,
- newTablePath)
- val dbName = oldTableIdentifier.getDatabaseName
- val tableName = oldTableIdentifier.getTableName
- sparkSession.catalog.refreshTable(TableIdentifier(tableName, Some(dbName)).quotedString)
- removeTableFromMetadata(dbName, tableName)
- CarbonMetadata.getInstance().loadTableMetadata(wrapperTableInfo)
- newTablePath
- }
-
/**
* This method will is used to remove the evolution entry in case of failure.
*
[09/14] carbondata git commit: [CARBONDATA-2014] Update table status
for failure only after first entry and IndexOutOfBound Fix
Posted by ja...@apache.org.
[CARBONDATA-2014] Update table status for failure only after first entry and IndexOutOfBound Fix
update table status for load failure only after first entry and before calling to update the table status for failure, check whether it is hive partition table in the same way as it is checked while updating in progress status to table status
This closes #1769
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/0adb32d9
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/0adb32d9
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/0adb32d9
Branch: refs/heads/carbonstore
Commit: 0adb32d900230ec3722faa3b06b569a5ce4d2729
Parents: 6123e7f
Author: akashrn5 <ak...@gmail.com>
Authored: Sat Jan 6 16:29:30 2018 +0530
Committer: manishgupta88 <to...@gmail.com>
Committed: Fri Jan 12 13:43:38 2018 +0530
----------------------------------------------------------------------
.../command/management/CarbonLoadDataCommand.scala | 10 ++++++++--
.../carbondata/processing/util/CarbonLoaderUtil.java | 8 --------
2 files changed, 8 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0adb32d9/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala
index 60adcd7..1828557 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala
@@ -166,6 +166,7 @@ case class CarbonLoadDataCommand(
// the Fact folder
LOGGER.info(s"Deleting stale folders if present for table $dbName.$tableName")
TableProcessingOperations.deletePartialLoadDataIfExist(table, false)
+ var isUpdateTableStatusRequired = false
try {
val operationContext = new OperationContext
val loadTablePreExecutionEvent: LoadTablePreExecutionEvent =
@@ -187,6 +188,7 @@ case class CarbonLoadDataCommand(
// add the start entry for the new load in the table status file
if (updateModel.isEmpty && !table.isHivePartitionTable) {
CarbonLoaderUtil.readAndUpdateLoadProgressInTableMeta(carbonLoadModel, isOverwriteTable)
+ isUpdateTableStatusRequired = true
}
if (isOverwriteTable) {
LOGGER.info(s"Overwrite of carbon table with $dbName.$tableName is in progress")
@@ -242,7 +244,9 @@ case class CarbonLoadDataCommand(
} catch {
case CausedBy(ex: NoRetryException) =>
// update the load entry in table status file for changing the status to marked for delete
- CarbonLoaderUtil.updateTableStatusForFailure(carbonLoadModel)
+ if (isUpdateTableStatusRequired) {
+ CarbonLoaderUtil.updateTableStatusForFailure(carbonLoadModel)
+ }
LOGGER.error(ex, s"Dataload failure for $dbName.$tableName")
throw new RuntimeException(s"Dataload failure for $dbName.$tableName, ${ex.getMessage}")
// In case of event related exception
@@ -251,7 +255,9 @@ case class CarbonLoadDataCommand(
case ex: Exception =>
LOGGER.error(ex)
// update the load entry in table status file for changing the status to marked for delete
- CarbonLoaderUtil.updateTableStatusForFailure(carbonLoadModel)
+ if (isUpdateTableStatusRequired) {
+ CarbonLoaderUtil.updateTableStatusForFailure(carbonLoadModel)
+ }
LOGGER.audit(s"Dataload failure for $dbName.$tableName. Please check the logs")
throw ex
} finally {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0adb32d9/processing/src/main/java/org/apache/carbondata/processing/util/CarbonLoaderUtil.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/util/CarbonLoaderUtil.java b/processing/src/main/java/org/apache/carbondata/processing/util/CarbonLoaderUtil.java
index de59982..1991017 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/util/CarbonLoaderUtil.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/util/CarbonLoaderUtil.java
@@ -212,21 +212,13 @@ public final class CarbonLoaderUtil {
// existing entry needs to be overwritten as the entry will exist with some
// intermediate status
int indexToOverwriteNewMetaEntry = 0;
- Boolean anyLoadInProgress = false;
for (LoadMetadataDetails entry : listOfLoadFolderDetails) {
- if (entry.getSegmentStatus().equals(SegmentStatus.INSERT_OVERWRITE_IN_PROGRESS)) {
- anyLoadInProgress = true;
- }
if (entry.getLoadName().equals(newMetaEntry.getLoadName())
&& entry.getLoadStartTime() == newMetaEntry.getLoadStartTime()) {
break;
}
indexToOverwriteNewMetaEntry++;
}
- if (listOfLoadFolderDetails.get(indexToOverwriteNewMetaEntry).getSegmentStatus() ==
- SegmentStatus.MARKED_FOR_DELETE && anyLoadInProgress) {
- throw new RuntimeException("It seems insert overwrite has been issued during load");
- }
if (insertOverwrite) {
for (LoadMetadataDetails entry : listOfLoadFolderDetails) {
if (entry.getSegmentStatus() != SegmentStatus.INSERT_OVERWRITE_IN_PROGRESS) {
[10/14] carbondata git commit: [CARBONDATA-1940][PreAgg] Fixed bug
for creation of preaggregate table with group by clause
Posted by ja...@apache.org.
[CARBONDATA-1940][PreAgg] Fixed bug for creation of preaggregate table with group by clause
1. Refactored code to create pre-aggregate table with group by clause
2. Added related test cases
This closes #1724
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/36b79826
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/36b79826
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/36b79826
Branch: refs/heads/carbonstore
Commit: 36b798262f271fe160c83498f6910944ca934962
Parents: 0adb32d
Author: Geetika Gupta <ge...@knoldus.in>
Authored: Tue Dec 26 17:29:38 2017 +0530
Committer: kunal642 <ku...@gmail.com>
Committed: Fri Jan 12 14:46:06 2018 +0530
----------------------------------------------------------------------
.../preaggregate/TestPreAggCreateCommand.scala | 12 ++++++++++++
.../preaggregate/TestPreAggregateLoad.scala | 16 ++++++++++++++++
.../TestPreAggregateTableSelection.scala | 11 +++++++++++
.../command/preaaggregate/PreAggregateUtil.scala | 17 +++++++++++++++++
.../spark/sql/hive/CarbonPreAggregateRules.scala | 6 +++++-
5 files changed, 61 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/36b79826/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggCreateCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggCreateCommand.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggCreateCommand.scala
index cb72732..5784bf2 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggCreateCommand.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggCreateCommand.scala
@@ -195,6 +195,18 @@ class TestPreAggCreateCommand extends QueryTest with BeforeAndAfterAll {
sql("drop datamap agg0 on table maintable")
}
+ test("test pre agg create table 20") {
+ sql("create datamap agg0 on table mainTable using 'preaggregate' as select column3, sum(column3),column5, sum(column5) from maintable group by column3,column5,column2")
+ val df = sql("select * from maintable_agg0")
+ val carbontable = getCarbontable(df.queryExecution.analyzed)
+ assert(carbontable.getAllMeasures.size()==2)
+ assert(carbontable.getAllDimensions.size()==3)
+ carbontable.getAllDimensions.asScala.foreach{ f =>
+ assert(!f.getEncoder.contains(Encoding.DICTIONARY))
+ }
+ sql("drop datamap agg0 on table maintable")
+ }
+
def getCarbontable(plan: LogicalPlan) : CarbonTable ={
var carbonTable : CarbonTable = null
http://git-wip-us.apache.org/repos/asf/carbondata/blob/36b79826/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateLoad.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateLoad.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateLoad.scala
index d794f32..fb5f81d 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateLoad.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateLoad.scala
@@ -238,4 +238,20 @@ class TestPreAggregateLoad extends QueryTest with BeforeAndAfterAll {
sql("drop table if exists maintable")
}
+ test("test load into preaggregate table having group by clause") {
+ sql("DROP TABLE IF EXISTS maintable")
+ sql(
+ """
+ | CREATE TABLE maintable(id int, name string, city string, age int)
+ | STORED BY 'org.apache.carbondata.format'
+ """.stripMargin)
+ sql(s"insert into maintable values(1, 'xyz', 'bengaluru', 26)")
+ sql(s"insert into maintable values(1, 'xyz', 'bengaluru', 26)")
+ sql("set carbon.input.segments.default.maintable=0")
+ sql(
+ s"""create datamap preagg_sum on table maintable using 'preaggregate' as select id, sum(age) from maintable group by id,name"""
+ .stripMargin)
+ checkAnswer(sql("select * from maintable_preagg_sum"), Row(1, 52, "xyz"))
+ }
+
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/36b79826/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateTableSelection.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateTableSelection.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateTableSelection.scala
index 322827e..2e0dcc4 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateTableSelection.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateTableSelection.scala
@@ -235,6 +235,17 @@ class TestPreAggregateTableSelection extends QueryTest with BeforeAndAfterAll {
preAggTableValidator(df.queryExecution.analyzed, "maintable_agg2")
}
+ test("test pre-agg table with group by condition") {
+ sql("drop table if exists grouptable")
+ sql("CREATE TABLE grouptable(id int, name string, city string, age string) STORED BY" +
+ " 'org.apache.carbondata.format' TBLPROPERTIES('dictionary_include'='name,age')")
+ sql(s"LOAD DATA LOCAL INPATH '$resourcesPath/measureinsertintotest.csv' into table grouptable")
+ sql("create datamap agg9 on table grouptable using 'preaggregate' as select sum(id) from grouptable group by city")
+ val df = sql("select sum(id) from grouptable group by city")
+ preAggTableValidator(df.queryExecution.analyzed, "grouptable_agg9")
+ checkAnswer(df, Seq(Row(3), Row(3), Row(4), Row(7)))
+ }
+
override def afterAll: Unit = {
sql("drop table if exists mainTable")
sql("drop table if exists lineitem")
http://git-wip-us.apache.org/repos/asf/carbondata/blob/36b79826/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateUtil.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateUtil.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateUtil.scala
index 86d0c6a..153c1a4 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateUtil.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateUtil.scala
@@ -155,6 +155,23 @@ object PreAggregateUtil {
throw new MalformedCarbonCommandException(s"Unsupported Select Statement:${
selectStmt } ")
}
+ groupByExp map {
+ case attr: AttributeReference =>
+ val columnRelation = getColumnRelation(
+ attr.name,
+ parentTableId,
+ parentTableName,
+ parentDatabaseName,
+ carbonTable)
+ fieldToDataMapFieldMap += createField(
+ attr.name,
+ attr.dataType,
+ parentTableName = parentTableName,
+ columnTableRelationList = Seq(columnRelation))
+ case _ =>
+ throw new MalformedCarbonCommandException(s"Unsupported Function in select Statement:${
+ selectStmt }")
+ }
fieldToDataMapFieldMap
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/36b79826/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonPreAggregateRules.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonPreAggregateRules.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonPreAggregateRules.scala
index bd8b4c6..8811a4e 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonPreAggregateRules.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonPreAggregateRules.scala
@@ -1008,7 +1008,7 @@ case class CarbonPreAggregateDataLoadingRules(sparkSession: SparkSession)
val validExpressionsMap = scala.collection.mutable.HashSet.empty[AggExpToColumnMappingModel]
val namedExpressionList = scala.collection.mutable.LinkedHashSet.empty[NamedExpression]
plan transform {
- case aggregate@Aggregate(_,
+ case aggregate@Aggregate(groupingExpressions,
aExp,
CarbonSubqueryAlias(_, logicalRelation: LogicalRelation))
if validateAggregateExpressions(aExp) &&
@@ -1066,6 +1066,10 @@ case class CarbonPreAggregateDataLoadingRules(sparkSession: SparkSession)
case alias@Alias(_: Expression, _) =>
namedExpressionList += alias
}
+ groupingExpressions foreach {
+ case namedExpr: NamedExpression => namedExpressionList += namedExpr
+ case _ => namedExpressionList
+ }
aggregate.copy(aggregateExpressions = namedExpressionList.toSeq)
case plan: LogicalPlan => plan
}
[13/14] carbondata git commit: [CARBONDATA-1978] Handled preaggregate
issues with hive metastore
Posted by ja...@apache.org.
[CARBONDATA-1978] Handled preaggregate issues with hive metastore
During creation of TableInfo from hivemetastore the DataMapSchemas and the columns DataTypes are not converted to the appropriate child classes due to which data types not supported exception is thrown
This closes #1758
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/df278312
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/df278312
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/df278312
Branch: refs/heads/carbonstore
Commit: df278312c32ee9e377eced1f7d70fcec9c9248ba
Parents: 6d77fdf
Author: kunal642 <ku...@gmail.com>
Authored: Wed Jan 3 19:17:53 2018 +0530
Committer: ravipesala <ra...@gmail.com>
Committed: Fri Jan 12 23:45:46 2018 +0530
----------------------------------------------------------------------
.../ThriftWrapperSchemaConverterImpl.java | 5 +-
.../core/metadata/schema/table/CarbonTable.java | 34 +++++++++++
.../carbondata/core/util/DataTypeUtil.java | 61 ++++++++++++++++++--
.../dictionary/client/DictionaryClientTest.java | 4 ++
...ncrementalColumnDictionaryGeneratorTest.java | 3 +
.../ServerDictionaryGeneratorTest.java | 3 +
.../generator/TableDictionaryGeneratorTest.java | 3 +
core/src/test/resources/sampleCSV.csv | 1 -
.../preaggregate/TestPreAggregateLoad.scala | 9 +++
.../testsuite/datamap/TestDataMapCommand.scala | 22 +++++++
10 files changed, 137 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/df278312/core/src/main/java/org/apache/carbondata/core/metadata/converter/ThriftWrapperSchemaConverterImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/metadata/converter/ThriftWrapperSchemaConverterImpl.java b/core/src/main/java/org/apache/carbondata/core/metadata/converter/ThriftWrapperSchemaConverterImpl.java
index 2b8cfa5..e9c5505 100644
--- a/core/src/main/java/org/apache/carbondata/core/metadata/converter/ThriftWrapperSchemaConverterImpl.java
+++ b/core/src/main/java/org/apache/carbondata/core/metadata/converter/ThriftWrapperSchemaConverterImpl.java
@@ -32,7 +32,6 @@ import org.apache.carbondata.core.metadata.schema.SchemaEvolutionEntry;
import org.apache.carbondata.core.metadata.schema.partition.PartitionType;
import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
import org.apache.carbondata.core.metadata.schema.table.DataMapSchema;
-import org.apache.carbondata.core.metadata.schema.table.DataMapSchemaFactory;
import org.apache.carbondata.core.metadata.schema.table.RelationIdentifier;
import org.apache.carbondata.core.metadata.schema.table.TableInfo;
import org.apache.carbondata.core.metadata.schema.table.TableSchema;
@@ -646,8 +645,8 @@ public class ThriftWrapperSchemaConverterImpl implements SchemaConverter {
@Override public DataMapSchema fromExternalToWrapperDataMapSchema(
org.apache.carbondata.format.DataMapSchema thriftDataMapSchema) {
- DataMapSchema childSchema = DataMapSchemaFactory.INSTANCE
- .getDataMapSchema(thriftDataMapSchema.getDataMapName(), thriftDataMapSchema.getClassName());
+ DataMapSchema childSchema = new DataMapSchema(thriftDataMapSchema.getDataMapName(),
+ thriftDataMapSchema.getClassName());
childSchema.setProperties(thriftDataMapSchema.getProperties());
if (null != thriftDataMapSchema.getRelationIdentifire()) {
RelationIdentifier relationIdentifier =
http://git-wip-us.apache.org/repos/asf/carbondata/blob/df278312/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/CarbonTable.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/CarbonTable.java b/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/CarbonTable.java
index 5e07fc0..e27b126 100644
--- a/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/CarbonTable.java
+++ b/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/CarbonTable.java
@@ -34,6 +34,7 @@ import org.apache.carbondata.core.metadata.schema.table.column.CarbonImplicitDim
import org.apache.carbondata.core.metadata.schema.table.column.CarbonMeasure;
import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema;
import org.apache.carbondata.core.util.CarbonUtil;
+import org.apache.carbondata.core.util.DataTypeUtil;
import org.apache.carbondata.core.util.path.CarbonTablePath;
/**
@@ -133,10 +134,43 @@ public class CarbonTable implements Serializable {
}
/**
+ * During creation of TableInfo from hivemetastore the DataMapSchemas and the columns
+ * DataTypes are not converted to the appropriate child classes.
+ *
+ * This method will cast the same to the appropriate classes
+ *
+ * @param tableInfo
+ */
+ private static void updateTableInfo(TableInfo tableInfo) {
+ List<DataMapSchema> dataMapSchemas = new ArrayList<>();
+ for (DataMapSchema dataMapSchema : tableInfo.getDataMapSchemaList()) {
+ DataMapSchema newDataMapSchema = DataMapSchemaFactory.INSTANCE
+ .getDataMapSchema(dataMapSchema.getDataMapName(), dataMapSchema.getClassName());
+ newDataMapSchema.setChildSchema(dataMapSchema.getChildSchema());
+ newDataMapSchema.setProperties(dataMapSchema.getProperties());
+ newDataMapSchema.setRelationIdentifier(dataMapSchema.getRelationIdentifier());
+ dataMapSchemas.add(newDataMapSchema);
+ }
+ tableInfo.setDataMapSchemaList(dataMapSchemas);
+ for (ColumnSchema columnSchema : tableInfo.getFactTable().getListOfColumns()) {
+ columnSchema.setDataType(DataTypeUtil.valueOf(columnSchema.getDataType(),
+ columnSchema.getPrecision(), columnSchema.getScale()));
+ }
+ if (tableInfo.getFactTable().getBucketingInfo() != null) {
+ for (ColumnSchema columnSchema : tableInfo.getFactTable()
+ .getBucketingInfo().getListOfColumns()) {
+ columnSchema.setDataType(DataTypeUtil.valueOf(columnSchema.getDataType(),
+ columnSchema.getPrecision(), columnSchema.getScale()));
+ }
+ }
+ }
+
+ /**
* @param tableInfo
*/
public static CarbonTable buildFromTableInfo(TableInfo tableInfo) {
CarbonTable table = new CarbonTable();
+ updateTableInfo(tableInfo);
table.tableInfo = tableInfo;
table.blockSize = tableInfo.getTableBlockSizeInMB();
table.tableLastUpdatedTime = tableInfo.getLastUpdatedTime();
http://git-wip-us.apache.org/repos/asf/carbondata/blob/df278312/core/src/main/java/org/apache/carbondata/core/util/DataTypeUtil.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/util/DataTypeUtil.java b/core/src/main/java/org/apache/carbondata/core/util/DataTypeUtil.java
index 5d188b5..6d224cf 100644
--- a/core/src/main/java/org/apache/carbondata/core/util/DataTypeUtil.java
+++ b/core/src/main/java/org/apache/carbondata/core/util/DataTypeUtil.java
@@ -735,10 +735,6 @@ public final class DataTypeUtil {
return converter;
}
- /**
- * @param name datatype string extracted from the json data
- * @return returns the datatype based on the input string from json to deserialize the tableInfo
- */
public static DataType valueOf(String name) {
if (DataTypes.STRING.getName().equalsIgnoreCase(name)) {
return DataTypes.STRING;
@@ -770,12 +766,69 @@ public final class DataTypeUtil {
return DataTypes.BYTE_ARRAY;
} else if (DataTypes.BYTE_ARRAY.getName().equalsIgnoreCase(name)) {
return DataTypes.BYTE_ARRAY;
+ } else if (name.equalsIgnoreCase("decimal")) {
+ return DataTypes.createDefaultDecimalType();
+ } else if (name.equalsIgnoreCase("array")) {
+ return DataTypes.createDefaultArrayType();
+ } else if (name.equalsIgnoreCase("struct")) {
+ return DataTypes.createDefaultStructType();
+ } else if (name.equalsIgnoreCase("map")) {
+ return DataTypes.createDefaultMapType();
} else {
throw new RuntimeException("create DataType with invalid name: " + name);
}
}
/**
+ * @param dataType extracted from the json data
+ * @return returns the datatype based on the input string from json to deserialize the tableInfo
+ */
+ public static DataType valueOf(DataType dataType, int precision, int scale) {
+ if (DataTypes.STRING.getName().equalsIgnoreCase(dataType.getName())) {
+ return DataTypes.STRING;
+ } else if (DataTypes.DATE.getName().equalsIgnoreCase(dataType.getName())) {
+ return DataTypes.DATE;
+ } else if (DataTypes.TIMESTAMP.getName().equalsIgnoreCase(dataType.getName())) {
+ return DataTypes.TIMESTAMP;
+ } else if (DataTypes.BOOLEAN.getName().equalsIgnoreCase(dataType.getName())) {
+ return DataTypes.BOOLEAN;
+ } else if (DataTypes.BYTE.getName().equalsIgnoreCase(dataType.getName())) {
+ return DataTypes.BYTE;
+ } else if (DataTypes.SHORT.getName().equalsIgnoreCase(dataType.getName())) {
+ return DataTypes.SHORT;
+ } else if (DataTypes.SHORT_INT.getName().equalsIgnoreCase(dataType.getName())) {
+ return DataTypes.SHORT_INT;
+ } else if (DataTypes.INT.getName().equalsIgnoreCase(dataType.getName())) {
+ return DataTypes.INT;
+ } else if (DataTypes.LONG.getName().equalsIgnoreCase(dataType.getName())) {
+ return DataTypes.LONG;
+ } else if (DataTypes.LEGACY_LONG.getName().equalsIgnoreCase(dataType.getName())) {
+ return DataTypes.LEGACY_LONG;
+ } else if (DataTypes.FLOAT.getName().equalsIgnoreCase(dataType.getName())) {
+ return DataTypes.FLOAT;
+ } else if (DataTypes.DOUBLE.getName().equalsIgnoreCase(dataType.getName())) {
+ return DataTypes.DOUBLE;
+ } else if (DataTypes.NULL.getName().equalsIgnoreCase(dataType.getName())) {
+ return DataTypes.NULL;
+ } else if (DataTypes.BYTE_ARRAY.getName().equalsIgnoreCase(dataType.getName())) {
+ return DataTypes.BYTE_ARRAY;
+ } else if (DataTypes.BYTE_ARRAY.getName().equalsIgnoreCase(dataType.getName())) {
+ return DataTypes.BYTE_ARRAY;
+ } else if (dataType.getName().equalsIgnoreCase("decimal")) {
+ return DataTypes.createDecimalType(precision, scale);
+ } else if (dataType.getName().equalsIgnoreCase("array")) {
+ return DataTypes.createDefaultArrayType();
+ } else if (dataType.getName().equalsIgnoreCase("struct")) {
+ return DataTypes.createDefaultStructType();
+ } else if (dataType.getName().equalsIgnoreCase("map")) {
+ return DataTypes.createDefaultMapType();
+ } else {
+ throw new RuntimeException(
+ "create DataType with invalid dataType.getName(): " + dataType.getName());
+ }
+ }
+
+ /**
* Method to type case the data based on modified data type. This method will used for
* retrieving the data after change in data type restructure operation
*
http://git-wip-us.apache.org/repos/asf/carbondata/blob/df278312/core/src/test/java/org/apache/carbondata/core/dictionary/client/DictionaryClientTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/carbondata/core/dictionary/client/DictionaryClientTest.java b/core/src/test/java/org/apache/carbondata/core/dictionary/client/DictionaryClientTest.java
index dbd29e5..4847177 100644
--- a/core/src/test/java/org/apache/carbondata/core/dictionary/client/DictionaryClientTest.java
+++ b/core/src/test/java/org/apache/carbondata/core/dictionary/client/DictionaryClientTest.java
@@ -28,6 +28,8 @@ import org.apache.carbondata.core.dictionary.generator.key.DictionaryMessageType
import org.apache.carbondata.core.dictionary.server.DictionaryServer;
import org.apache.carbondata.core.dictionary.server.NonSecureDictionaryServer;
import org.apache.carbondata.core.metadata.CarbonMetadata;
+import org.apache.carbondata.core.metadata.datatype.DataType;
+import org.apache.carbondata.core.metadata.datatype.DataTypes;
import org.apache.carbondata.core.metadata.encoder.Encoding;
import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
import org.apache.carbondata.core.metadata.schema.table.TableInfo;
@@ -68,11 +70,13 @@ public class DictionaryClientTest {
empColumnSchema.setColumnName("empNameCol");
empColumnSchema.setColumnUniqueId("empNameCol");
empColumnSchema.setDimensionColumn(true);
+ empColumnSchema.setDataType(DataTypes.STRING);
empColumnSchema.setEncodingList(Arrays.asList(Encoding.DICTIONARY));
empDimension = new CarbonDimension(empColumnSchema, 0, 0, 0, 0, 0);
ageColumnSchema = new ColumnSchema();
ageColumnSchema.setColumnName("ageNameCol");
+ ageColumnSchema.setDataType(DataTypes.SHORT_INT);
ageColumnSchema.setColumnUniqueId("ageNameCol");
ageColumnSchema.setDimensionColumn(true);
ageColumnSchema.setEncodingList(Arrays.asList(Encoding.DICTIONARY));
http://git-wip-us.apache.org/repos/asf/carbondata/blob/df278312/core/src/test/java/org/apache/carbondata/core/dictionary/generator/IncrementalColumnDictionaryGeneratorTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/carbondata/core/dictionary/generator/IncrementalColumnDictionaryGeneratorTest.java b/core/src/test/java/org/apache/carbondata/core/dictionary/generator/IncrementalColumnDictionaryGeneratorTest.java
index 414a350..2d31fa5 100644
--- a/core/src/test/java/org/apache/carbondata/core/dictionary/generator/IncrementalColumnDictionaryGeneratorTest.java
+++ b/core/src/test/java/org/apache/carbondata/core/dictionary/generator/IncrementalColumnDictionaryGeneratorTest.java
@@ -22,6 +22,7 @@ import java.util.Arrays;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.metadata.CarbonMetadata;
+import org.apache.carbondata.core.metadata.datatype.DataTypes;
import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
import org.apache.carbondata.core.metadata.schema.table.TableInfo;
import org.apache.carbondata.core.metadata.schema.table.TableSchema;
@@ -50,6 +51,7 @@ public class IncrementalColumnDictionaryGeneratorTest {
.addProperty(CarbonCommonConstants.CARBON_MAX_DRIVER_LRU_CACHE_SIZE, "10");
ColumnSchema columnSchema = new ColumnSchema();
columnSchema.setColumnName("empName");
+ columnSchema.setDataType(DataTypes.STRING);
TableSchema tableSchema = new TableSchema();
tableSchema.setTableName("TestTable");
tableSchema.setListOfColumns(Arrays.asList(columnSchema));
@@ -131,6 +133,7 @@ public class IncrementalColumnDictionaryGeneratorTest {
//Create required column schema
ColumnSchema columnSchema = new ColumnSchema();
columnSchema.setColumnName("empNameCol");
+ columnSchema.setDataType(DataTypes.STRING);
columnSchema.setColumnUniqueId("empNameCol");
CarbonDimension carbonDimension = new CarbonDimension(columnSchema, 0, 0, 0, 0, 0);
http://git-wip-us.apache.org/repos/asf/carbondata/blob/df278312/core/src/test/java/org/apache/carbondata/core/dictionary/generator/ServerDictionaryGeneratorTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/carbondata/core/dictionary/generator/ServerDictionaryGeneratorTest.java b/core/src/test/java/org/apache/carbondata/core/dictionary/generator/ServerDictionaryGeneratorTest.java
index 95d73a9..2564ead 100644
--- a/core/src/test/java/org/apache/carbondata/core/dictionary/generator/ServerDictionaryGeneratorTest.java
+++ b/core/src/test/java/org/apache/carbondata/core/dictionary/generator/ServerDictionaryGeneratorTest.java
@@ -23,6 +23,7 @@ import java.util.Arrays;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.dictionary.generator.key.DictionaryMessage;
import org.apache.carbondata.core.metadata.CarbonMetadata;
+import org.apache.carbondata.core.metadata.datatype.DataTypes;
import org.apache.carbondata.core.metadata.encoder.Encoding;
import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
import org.apache.carbondata.core.metadata.schema.table.TableInfo;
@@ -60,6 +61,7 @@ public class ServerDictionaryGeneratorTest {
// Create two column schemas and dimensions for the table
empColumnSchema = new ColumnSchema();
empColumnSchema.setColumnName("empNameCol");
+ empColumnSchema.setDataType(DataTypes.STRING);
empColumnSchema.setColumnUniqueId("empNameCol");
empColumnSchema.setDimensionColumn(true);
empColumnSchema.setEncodingList(Arrays.asList(Encoding.DICTIONARY));
@@ -68,6 +70,7 @@ public class ServerDictionaryGeneratorTest {
ageColumnSchema = new ColumnSchema();
ageColumnSchema.setColumnName("empNameCol");
ageColumnSchema.setColumnUniqueId("empNameCol");
+ ageColumnSchema.setDataType(DataTypes.SHORT_INT);
ageColumnSchema.setDimensionColumn(true);
ageColumnSchema.setEncodingList(Arrays.asList(Encoding.DICTIONARY));
ageDimension = new CarbonDimension(ageColumnSchema, 0, 0, 0, 0, 0);
http://git-wip-us.apache.org/repos/asf/carbondata/blob/df278312/core/src/test/java/org/apache/carbondata/core/dictionary/generator/TableDictionaryGeneratorTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/carbondata/core/dictionary/generator/TableDictionaryGeneratorTest.java b/core/src/test/java/org/apache/carbondata/core/dictionary/generator/TableDictionaryGeneratorTest.java
index 1ea77f8..e40bdad 100644
--- a/core/src/test/java/org/apache/carbondata/core/dictionary/generator/TableDictionaryGeneratorTest.java
+++ b/core/src/test/java/org/apache/carbondata/core/dictionary/generator/TableDictionaryGeneratorTest.java
@@ -23,6 +23,7 @@ import java.util.Arrays;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.dictionary.generator.key.DictionaryMessage;
import org.apache.carbondata.core.metadata.CarbonMetadata;
+import org.apache.carbondata.core.metadata.datatype.DataTypes;
import org.apache.carbondata.core.metadata.encoder.Encoding;
import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
import org.apache.carbondata.core.metadata.schema.table.TableInfo;
@@ -62,10 +63,12 @@ public class TableDictionaryGeneratorTest {
empColumnSchema.setColumnName("empNameCol");
empColumnSchema.setColumnUniqueId("empNameCol");
empColumnSchema.setDimensionColumn(true);
+ empColumnSchema.setDataType(DataTypes.STRING);
empColumnSchema.setEncodingList(Arrays.asList(Encoding.DICTIONARY));
empDimension = new CarbonDimension(empColumnSchema, 0, 0, 0, 0, 0);
ageColumnSchema = new ColumnSchema();
+ ageColumnSchema.setDataType(DataTypes.SHORT_INT);
ageColumnSchema.setColumnName("ageNameCol");
ageColumnSchema.setColumnUniqueId("ageNameCol");
ageColumnSchema.setDimensionColumn(true);
http://git-wip-us.apache.org/repos/asf/carbondata/blob/df278312/core/src/test/resources/sampleCSV.csv
----------------------------------------------------------------------
diff --git a/core/src/test/resources/sampleCSV.csv b/core/src/test/resources/sampleCSV.csv
deleted file mode 100644
index 79dfd50..0000000
--- a/core/src/test/resources/sampleCSV.csv
+++ /dev/null
@@ -1 +0,0 @@
-id,name
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/carbondata/blob/df278312/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateLoad.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateLoad.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateLoad.scala
index fb5f81d..3d0e9b9 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateLoad.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateLoad.scala
@@ -168,6 +168,15 @@ class TestPreAggregateLoad extends QueryTest with BeforeAndAfterAll {
}
test("test to check if exception is thrown for direct load on pre-aggregate table") {
+ sql("drop table if exists maintable")
+ sql(
+ """
+ | CREATE TABLE maintable(id int, name string, city string, age int)
+ | STORED BY 'org.apache.carbondata.format' TBLPROPERTIES('dictionary_include'='id')
+ """.stripMargin)
+ sql(
+ s"""create datamap preagg_sum on table maintable using 'preaggregate' as select id,sum(age) from maintable group by id"""
+ .stripMargin)
assert(intercept[RuntimeException] {
sql(s"insert into maintable_preagg_sum values(1, 30)")
}.getMessage.equalsIgnoreCase("Cannot insert/load data directly into pre-aggregate table"))
http://git-wip-us.apache.org/repos/asf/carbondata/blob/df278312/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/TestDataMapCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/TestDataMapCommand.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/TestDataMapCommand.scala
index c38e6cf..d0a342b 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/TestDataMapCommand.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/TestDataMapCommand.scala
@@ -17,6 +17,7 @@
package org.apache.carbondata.spark.testsuite.datamap
+import org.apache.spark.sql.Row
import org.apache.spark.sql.test.util.QueryTest
import org.scalatest.BeforeAndAfterAll
@@ -26,6 +27,8 @@ import org.apache.carbondata.core.util.CarbonProperties
class TestDataMapCommand extends QueryTest with BeforeAndAfterAll {
+ val testData = s"$resourcesPath/sample.csv"
+
override def beforeAll {
sql("drop table if exists datamaptest")
sql("drop table if exists datamapshowtest")
@@ -188,8 +191,27 @@ class TestDataMapCommand extends QueryTest with BeforeAndAfterAll {
checkExistence(frame, true, "datamap2", "(NA)", "new.class")
}
+ test("test if preaggregate load is successfull for hivemetastore") {
+ CarbonProperties.getInstance().addProperty(CarbonCommonConstants.ENABLE_HIVE_SCHEMA_META_STORE, "true")
+ sql("DROP TABLE IF EXISTS maintable")
+ sql(
+ """
+ | CREATE TABLE maintable(id int, name string, city string, age int)
+ | STORED BY 'org.apache.carbondata.format'
+ """.stripMargin)
+ sql(
+ s"""create datamap preagg_sum on table maintable using 'preaggregate' as select id,sum(age) from maintable group by id"""
+ .stripMargin)
+ sql(s"LOAD DATA LOCAL INPATH '$testData' into table maintable")
+ checkAnswer(sql(s"select * from maintable_preagg_sum"),
+ Seq(Row(1, 31), Row(2, 27), Row(3, 70), Row(4, 55)))
+ }
+
override def afterAll {
+ sql("DROP TABLE IF EXISTS maintable")
+ CarbonProperties.getInstance().addProperty(CarbonCommonConstants.ENABLE_HIVE_SCHEMA_META_STORE,
+ CarbonCommonConstants.ENABLE_HIVE_SCHEMA_META_STORE_DEFAULT)
sql("drop table if exists datamaptest")
sql("drop table if exists datamapshowtest")
}
[08/14] carbondata git commit: [CARBONDATA-1981][Build] Fix compile
error in windows env
Posted by ja...@apache.org.
[CARBONDATA-1981][Build] Fix compile error in windows env
Fix compile error in windows env
This closes #1762
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/6123e7f3
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/6123e7f3
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/6123e7f3
Branch: refs/heads/carbonstore
Commit: 6123e7f383053d6222ea7870f3028709df4b8db6
Parents: f557545
Author: xuchuanyin <xu...@hust.edu.cn>
Authored: Fri Jan 5 15:22:01 2018 +0800
Committer: chenliang613 <ch...@huawei.com>
Committed: Thu Jan 11 18:43:39 2018 +0800
----------------------------------------------------------------------
build/README.md | 2 ++
build/carbondata-build-info | 36 ----------------------------------
build/carbondata-build-info.bat | 38 ++++++++++++++++++++++++++++++++++++
build/carbondata-build-info.sh | 36 ++++++++++++++++++++++++++++++++++
core/pom.xml | 22 ++++++++++-----------
pom.xml | 7 +++++++
6 files changed, 93 insertions(+), 48 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/6123e7f3/build/README.md
----------------------------------------------------------------------
diff --git a/build/README.md b/build/README.md
index 65e58d8..028c03a 100644
--- a/build/README.md
+++ b/build/README.md
@@ -31,6 +31,8 @@ mvn -DskipTests -Pspark-2.1 -Dspark.version=2.1.0 clean package
mvn -DskipTests -Pspark-2.2 -Dspark.version=2.2.1 clean package
```
+Note: If you are working in Windows environment, remember to add `-Pwindows` while building the project.
+
## For contributors : To build the format code after any changes, please follow the below command.
Note:Need install Apache Thrift 0.9.3
```
http://git-wip-us.apache.org/repos/asf/carbondata/blob/6123e7f3/build/carbondata-build-info
----------------------------------------------------------------------
diff --git a/build/carbondata-build-info b/build/carbondata-build-info
deleted file mode 100755
index 0d619fc..0000000
--- a/build/carbondata-build-info
+++ /dev/null
@@ -1,36 +0,0 @@
-#!/usr/bin/env bash
-
-#
-# 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.
-#
-
-# This script generates the build info for carbondata and places it into the carbondata-version-info.properties file.
-# Arguments:
-# build_tgt_directory - The target directory where properties file would be created. [./core/target/extra-resources]
-# carbondata_version - The current version of carbondata
-
-RESOURCE_DIR="$1"
-mkdir -p "$RESOURCE_DIR"
-CARBONDATA_BUILD_INFO="${RESOURCE_DIR}"/carbondata-version-info.properties
-
-echo_build_properties() {
- echo version=$1
- echo revision=$(git rev-parse HEAD)
- echo branch=$(git rev-parse --abbrev-ref HEAD)
- echo date=$(date -u +%Y-%m-%dT%H:%M:%SZ)
-}
-
-echo_build_properties $2 > "$CARBONDATA_BUILD_INFO"
http://git-wip-us.apache.org/repos/asf/carbondata/blob/6123e7f3/build/carbondata-build-info.bat
----------------------------------------------------------------------
diff --git a/build/carbondata-build-info.bat b/build/carbondata-build-info.bat
new file mode 100644
index 0000000..c1b8b27
--- /dev/null
+++ b/build/carbondata-build-info.bat
@@ -0,0 +1,38 @@
+@echo off
+
+rem
+rem Licensed to the Apache Software Foundation (ASF) under one or more
+rem contributor license agreements. See the NOTICE file distributed with
+rem this work for additional information regarding copyright ownership.
+rem The ASF licenses this file to You under the Apache License, Version 2.0
+rem (the "License"); you may not use this file except in compliance with
+rem the License. You may obtain a copy of the License at
+rem
+rem http://www.apache.org/licenses/LICENSE-2.0
+rem
+rem Unless required by applicable law or agreed to in writing, software
+rem distributed under the License is distributed on an "AS IS" BASIS,
+rem WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+rem See the License for the specific language governing permissions and
+rem limitations under the License.
+rem
+
+rem This script generates the build info for carbondata and places it into the carbondata-version-info.properties file.
+rem Arguments:
+rem build_tgt_directory - The target directory where properties file would be created. [./core/target/extra-resources]
+rem carbondata_version - The current version of carbondata
+
+set RESOURCE_DIR=%1
+md "%RESOURCE_DIR%"
+set CARBONDATA_BUILD_INFO=%RESOURCE_DIR%/carbondata-version-info.properties
+
+echo version=%2> %CARBONDATA_BUILD_INFO%
+
+for /f %%i in ('git rev-parse HEAD') do set revision=%%i
+echo revision=%revision%>> %CARBONDATA_BUILD_INFO%
+
+for /f %%i in ('git rev-parse --abbrev-ref HEAD') do set branch=%%i
+echo branch=%branch%>> %CARBONDATA_BUILD_INFO%
+
+for /f "delims=," %%i in ('echo %date:/=-%T%time%Z') do set day=%%i
+echo date=%day%>> %CARBONDATA_BUILD_INFO%
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/carbondata/blob/6123e7f3/build/carbondata-build-info.sh
----------------------------------------------------------------------
diff --git a/build/carbondata-build-info.sh b/build/carbondata-build-info.sh
new file mode 100755
index 0000000..0d619fc
--- /dev/null
+++ b/build/carbondata-build-info.sh
@@ -0,0 +1,36 @@
+#!/usr/bin/env bash
+
+#
+# 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.
+#
+
+# This script generates the build info for carbondata and places it into the carbondata-version-info.properties file.
+# Arguments:
+# build_tgt_directory - The target directory where properties file would be created. [./core/target/extra-resources]
+# carbondata_version - The current version of carbondata
+
+RESOURCE_DIR="$1"
+mkdir -p "$RESOURCE_DIR"
+CARBONDATA_BUILD_INFO="${RESOURCE_DIR}"/carbondata-version-info.properties
+
+echo_build_properties() {
+ echo version=$1
+ echo revision=$(git rev-parse HEAD)
+ echo branch=$(git rev-parse --abbrev-ref HEAD)
+ echo date=$(date -u +%Y-%m-%dT%H:%M:%SZ)
+}
+
+echo_build_properties $2 > "$CARBONDATA_BUILD_INFO"
http://git-wip-us.apache.org/repos/asf/carbondata/blob/6123e7f3/core/pom.xml
----------------------------------------------------------------------
diff --git a/core/pom.xml b/core/pom.xml
index ebcf0ee..f874615 100644
--- a/core/pom.xml
+++ b/core/pom.xml
@@ -114,24 +114,22 @@
</resources>
<plugins>
<plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-antrun-plugin</artifactId>
- <version>1.8</version>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>exec-maven-plugin</artifactId>
+ <version>1.2.1</version>
<executions>
<execution>
<phase>generate-resources</phase>
+ <!-- Execute the shell script to generate the CarbonData build information. -->
<configuration>
- <!-- Execute the shell script to generate the CarbonData build information. -->
- <target>
- <exec executable="bash">
- <arg value="${project.basedir}/../build/carbondata-build-info"/>
- <arg value="${project.build.directory}/extra-resources"/>
- <arg value="${project.version}"/>
- </exec>
- </target>
+ <executable>${project.basedir}/../build/carbondata-build-info${script.exetension}</executable>
+ <arguments>
+ <argument>${project.build.directory}/extra-resources</argument>
+ <argument>${project.version}</argument>
+ </arguments>
</configuration>
<goals>
- <goal>run</goal>
+ <goal>exec</goal>
</goals>
</execution>
</executions>
http://git-wip-us.apache.org/repos/asf/carbondata/blob/6123e7f3/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 0cbfc7b..07fd946 100644
--- a/pom.xml
+++ b/pom.xml
@@ -117,6 +117,7 @@
<spark.master.url>local[2]</spark.master.url>
<hdfs.url>local</hdfs.url>
<suite.name>org.apache.carbondata.cluster.sdv.suite.SDVSuites</suite.name>
+ <script.exetension>.sh</script.exetension>
</properties>
<repositories>
@@ -603,6 +604,12 @@
</plugins>
</build>
</profile>
+ <profile>
+ <id>windows</id>
+ <properties>
+ <script.exetension>.bat</script.exetension>
+ </properties>
+ </profile>
</profiles>
</project>
[06/14] carbondata git commit: [CARBONDATA-1993] Removed unused
carbon properties high.cardinality.identify
Posted by ja...@apache.org.
[CARBONDATA-1993] Removed unused carbon properties high.cardinality.identify
This closes #1775
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/1012c627
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/1012c627
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/1012c627
Branch: refs/heads/carbonstore
Commit: 1012c62742e18d7e721565a7c362c090d0888864
Parents: 3eb1e2b
Author: Pawan Malwal <pm...@gmail.com>
Authored: Mon Jan 8 16:00:47 2018 +0530
Committer: Jacky Li <ja...@qq.com>
Committed: Thu Jan 11 09:37:17 2018 +0800
----------------------------------------------------------------------
conf/carbon.properties.template | 4 ----
docs/configuration-parameters.md | 2 --
docs/data-management-on-carbondata.md | 1 -
3 files changed, 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/1012c627/conf/carbon.properties.template
----------------------------------------------------------------------
diff --git a/conf/carbon.properties.template b/conf/carbon.properties.template
index de8a4ee..8c1e458 100644
--- a/conf/carbon.properties.template
+++ b/conf/carbon.properties.template
@@ -107,10 +107,6 @@ carbon.enable.quick.filter=false
##Min max is feature added to enhance query performance. To disable this feature, make it false.
#carbon.enableMinMax=true
######## Global Dictionary Configurations ########
-##To enable/disable identify high cardinality during first data loading
-#high.cardinality.identify.enable=true
-##threshold to identify whether high cardinality column
-#high.cardinality.threshold=1000000
##Percentage to identify whether column cardinality is more than configured percent of total row count
#high.cardinality.row.count.percentage=80
##The property to set the date to be considered as start date for calculating the timestamp.
http://git-wip-us.apache.org/repos/asf/carbondata/blob/1012c627/docs/configuration-parameters.md
----------------------------------------------------------------------
diff --git a/docs/configuration-parameters.md b/docs/configuration-parameters.md
index 5abfdfd..05aede7 100644
--- a/docs/configuration-parameters.md
+++ b/docs/configuration-parameters.md
@@ -130,8 +130,6 @@ This section provides the details of all the configurations required for CarbonD
| Parameter | Default Value | Description |
|---------------------------------------|---------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| high.cardinality.identify.enable | true | If the parameter is true, the high cardinality columns of the dictionary code are automatically recognized and these columns will not be used as global dictionary encoding. If the parameter is false, all dictionary encoding columns are used as dictionary encoding. The high cardinality column must meet the following requirements: value of cardinality > configured value of high.cardinality. <b> Note: </b> If SINGLE_PASS is used during data load, then this property will be disabled.|
-| high.cardinality.threshold | 1000000 | It is a threshold to identify high cardinality of the columns.If the value of columns' cardinality > the configured value, then the columns are excluded from dictionary encoding. |
| carbon.cutOffTimestamp | 1970-01-01 05:30:00 | Sets the start date for calculating the timestamp. Java counts the number of milliseconds from start of "1970-01-01 00:00:00". This property is used to customize the start of position. For example "2000-01-01 00:00:00". The date must be in the form "carbon.timestamp.format". |
| carbon.timegranularity | SECOND | The property used to set the data granularity level DAY, HOUR, MINUTE, or SECOND. |
http://git-wip-us.apache.org/repos/asf/carbondata/blob/1012c627/docs/data-management-on-carbondata.md
----------------------------------------------------------------------
diff --git a/docs/data-management-on-carbondata.md b/docs/data-management-on-carbondata.md
index 39acfbf..0ca9289 100644
--- a/docs/data-management-on-carbondata.md
+++ b/docs/data-management-on-carbondata.md
@@ -372,7 +372,6 @@ This tutorial is going to introduce all commands and data operations on CarbonDa
NOTE:
* If this option is set to TRUE then data loading will take less time.
* If this option is set to some invalid value other than TRUE or FALSE then it uses the default value.
- * If this option is set to TRUE, then high.cardinality.identify.enable property will be disabled during data load.
Example:
[07/14] carbondata git commit: [CARBONDATA-1592] Added analysis
exception to handle event exceptions
Posted by ja...@apache.org.
[CARBONDATA-1592] Added analysis exception to handle event exceptions
This closes #1788
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/f557545d
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/f557545d
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/f557545d
Branch: refs/heads/carbonstore
Commit: f557545d19565ffa0116e9689253f1aa5de50381
Parents: 1012c62
Author: Manohar <ma...@gmail.com>
Authored: Wed Jan 10 20:07:33 2018 +0530
Committer: ravipesala <ra...@gmail.com>
Committed: Thu Jan 11 16:06:47 2018 +0530
----------------------------------------------------------------------
.../org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala | 5 ++++-
1 file changed, 4 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/f557545d/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala b/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
index 7982071..b03b6fa 100644
--- a/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
+++ b/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
@@ -37,7 +37,7 @@ import org.apache.hadoop.mapreduce.lib.input.{FileInputFormat, FileSplit}
import org.apache.spark.{SparkEnv, SparkException, TaskContext}
import org.apache.spark.deploy.SparkHadoopUtil
import org.apache.spark.rdd.{DataLoadCoalescedRDD, DataLoadPartitionCoalescer, NewHadoopRDD, RDD}
-import org.apache.spark.sql.{CarbonEnv, DataFrame, Row, SQLContext}
+import org.apache.spark.sql.{AnalysisException, CarbonEnv, DataFrame, Row, SQLContext}
import org.apache.spark.sql.execution.command.{CompactionModel, ExecutionErrors, UpdateTableModel}
import org.apache.spark.sql.hive.DistributionUtil
import org.apache.spark.sql.optimizer.CarbonFilters
@@ -417,6 +417,9 @@ object CarbonDataRDDFactory {
.trimErrorMessage(sparkException.getCause.getMessage)
errorMessage = errorMessage + " : " + executorMessage
}
+ case aex: AnalysisException =>
+ LOGGER.error(aex.getMessage())
+ throw aex
case _ =>
if (ex.getCause != null) {
executorMessage = ex.getCause.getMessage