You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ma...@apache.org on 2018/04/11 05:41:51 UTC
carbondata git commit: [CARBONDATA-2317] Concurrent datamap with same
name and schema creation throws exception
Repository: carbondata
Updated Branches:
refs/heads/master 9ca9b6d0c -> 4cbd5cdf2
[CARBONDATA-2317] Concurrent datamap with same name and schema creation throws exception
Concurrent datamap with same name and schema creation throws exception
This closes #2143
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/4cbd5cdf
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/4cbd5cdf
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/4cbd5cdf
Branch: refs/heads/master
Commit: 4cbd5cdf20341f1ae07c7cbb8275c1e333856cae
Parents: 9ca9b6d
Author: rahulforallp <ra...@knoldus.in>
Authored: Fri Apr 6 15:17:54 2018 +0530
Committer: manishgupta88 <to...@gmail.com>
Committed: Wed Apr 11 11:15:15 2018 +0530
----------------------------------------------------------------------
.../preaggregate/TestPreAggCreateCommand.scala | 44 ++++++++++++++++++++
.../table/CarbonCreateTableCommand.scala | 31 +++++++-------
2 files changed, 61 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/4cbd5cdf/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 e546fe8..7cb1adf 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
@@ -17,7 +17,14 @@
package org.apache.carbondata.integration.spark.testsuite.preaggregate
+import java.util
+import java.util.concurrent.{Callable, ExecutorService, Executors, TimeUnit}
+
import scala.collection.JavaConverters._
+import scala.concurrent.ExecutionContext.Implicits.global
+import scala.concurrent.{Await, Future}
+import scala.concurrent.duration.Duration
+import scala.util.{Failure, Success}
import org.apache.spark.sql.{AnalysisException, CarbonDatasourceHadoopRelation, Row}
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
@@ -429,6 +436,43 @@ class TestPreAggCreateCommand extends QueryTest with BeforeAndAfterAll {
}
}
+ test("test creation of multiple preaggregate of same name concurrently ") {
+ sql("DROP TABLE IF EXISTS tbl_concurr")
+ sql(
+ "create table if not exists tbl_concurr(imei string,age int,mac string ,prodate timestamp," +
+ "update timestamp,gamepoint double,contrid double) stored by 'carbondata' ")
+
+ var executorService: ExecutorService = Executors.newCachedThreadPool()
+ val tasks = new util.ArrayList[Callable[String]]()
+ var i = 0
+ val count = 5
+ while (i < count) {
+ tasks
+ .add(new QueryTask(
+ s"""create datamap agg_concu1 on table tbl_concurr using
+ |'preaggregate' as select prodate, mac from tbl_concurr group by prodate,mac"""
+ .stripMargin))
+ i = i + 1
+ }
+ executorService.invokeAll(tasks)
+
+ checkExistence(sql("show tables"), true, "agg_concu1", "tbl_concurr")
+ executorService.shutdown()
+ }
+
+ class QueryTask(query: String) extends Callable[String] {
+ override def call(): String = {
+ var result = "SUCCESS"
+ try {
+ sql(query).collect()
+ } catch {
+ case exception: Exception => LOGGER.error(exception.getMessage)
+ }
+ result
+ }
+ }
+
+
def getCarbonTable(plan: LogicalPlan) : CarbonTable ={
var carbonTable : CarbonTable = null
plan.transform {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/4cbd5cdf/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonCreateTableCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonCreateTableCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonCreateTableCommand.scala
index 65c6269..6266c53 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonCreateTableCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonCreateTableCommand.scala
@@ -121,21 +121,24 @@ case class CarbonCreateTableCommand(
// isVisible property is added to hive table properties to differentiate between main
// table and datamaps(like preaggregate). It is false only for datamaps. This is added
// to improve the show tables performance when filtering the datamaps from main tables
- sparkSession.sql(
- s"""CREATE TABLE $dbName.$tableName
- |(${ rawSchema })
- |USING org.apache.spark.sql.CarbonSource
- |OPTIONS (
- | tableName "$tableName",
- | dbName "$dbName",
- | tablePath "$tablePath",
- | path "$tablePath",
- | isExternal "$isExternal",
- | isUnManaged "$isUnmanaged",
- | isVisible "$isVisible"
- | $carbonSchemaString)
- | $partitionString
+ // synchronized to prevent concurrently creation of table with same name
+ CarbonCreateTableCommand.synchronized {
+ sparkSession.sql(
+ s"""CREATE TABLE $dbName.$tableName
+ |(${ rawSchema })
+ |USING org.apache.spark.sql.CarbonSource
+ |OPTIONS (
+ | tableName "$tableName",
+ | dbName "$dbName",
+ | tablePath "$tablePath",
+ | path "$tablePath",
+ | isExternal "$isExternal",
+ | isUnManaged "$isUnmanaged",
+ | isVisible "$isVisible"
+ | $carbonSchemaString)
+ | $partitionString
""".stripMargin)
+ }
} catch {
case e: AnalysisException => throw e
case e: Exception =>