You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ra...@apache.org on 2016/11/07 11:51:54 UTC
[1/5] incubator-carbondata git commit: fixed block loading issue in
case of blocklet distribution
Repository: incubator-carbondata
Updated Branches:
refs/heads/branch-0.2 5a97fc795 -> e0261831d
fixed block loading issue in case of blocklet distribution
Project: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/commit/03066142
Tree: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/tree/03066142
Diff: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/diff/03066142
Branch: refs/heads/branch-0.2
Commit: 03066142b3c5600b8a69f91a67d71d46fde7d0b4
Parents: 5a97fc7
Author: kumarvishal <ku...@gmail.com>
Authored: Wed Nov 2 22:32:42 2016 +0530
Committer: ravipesala <ra...@gmail.com>
Committed: Mon Nov 7 16:55:45 2016 +0530
----------------------------------------------------------------------
.../core/carbon/datastore/BlockIndexStore.java | 25 ++++++++++++++------
.../core/carbon/datastore/block/BlockInfo.java | 2 +-
.../core/constants/CarbonCommonConstants.java | 2 +-
3 files changed, 20 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/03066142/core/src/main/java/org/apache/carbondata/core/carbon/datastore/BlockIndexStore.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/carbon/datastore/BlockIndexStore.java b/core/src/main/java/org/apache/carbondata/core/carbon/datastore/BlockIndexStore.java
index bbebda0..34c2983 100644
--- a/core/src/main/java/org/apache/carbondata/core/carbon/datastore/BlockIndexStore.java
+++ b/core/src/main/java/org/apache/carbondata/core/carbon/datastore/BlockIndexStore.java
@@ -80,6 +80,13 @@ public class BlockIndexStore {
*/
private Map<AbsoluteTableIdentifier, Object> tableLockMap;
+ /**
+ * block info to future task mapping
+ * useful when blocklet distribution is enabled and
+ * same block is loaded by multiple thread
+ */
+ private Map<BlockInfo, Future<AbstractIndex>> mapOfBlockInfoToFuture;
+
private BlockIndexStore() {
tableBlocksMap = new ConcurrentHashMap<AbsoluteTableIdentifier, Map<BlockInfo, AbstractIndex>>(
CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
@@ -87,6 +94,7 @@ public class BlockIndexStore {
CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
blockInfoLock = new ConcurrentHashMap<BlockInfo, Object>();
segmentIdToBlockListMap = new ConcurrentHashMap<>();
+ mapOfBlockInfoToFuture = new ConcurrentHashMap<>();
}
/**
@@ -137,7 +145,6 @@ public class BlockIndexStore {
blockInfosNeedToLoad = fillSegmentIdToTableInfoMap(tableBlocksInfos, absoluteTableIdentifier);
}
AbstractIndex tableBlock = null;
- List<Future<AbstractIndex>> blocksList = new ArrayList<Future<AbstractIndex>>();
int counter = -1;
for (BlockInfo blockInfo : blockInfosNeedToLoad) {
counter++;
@@ -173,7 +180,12 @@ public class BlockIndexStore {
tableBlock = tableBlockMapTemp.get(blockInfo);
// if still block is not present then load the block
if (null == tableBlock) {
- blocksList.add(executor.submit(new BlockLoaderThread(blockInfo, tableBlockMapTemp)));
+ if (null == mapOfBlockInfoToFuture.get(blockInfo)) {
+ mapOfBlockInfoToFuture.put(blockInfo, executor
+ .submit(new BlockLoaderThread(blockInfo, tableBlockMapTemp)));
+ }
+ } else {
+ loadedBlock[counter] = tableBlock;
}
}
} else {
@@ -190,7 +202,7 @@ public class BlockIndexStore {
throw new IndexBuilderException(e);
}
// fill the block which were not loaded before to loaded blocks array
- fillLoadedBlocks(loadedBlock, blocksList);
+ fillLoadedBlocks(loadedBlock, blockInfosNeedToLoad);
return Arrays.asList(loadedBlock);
}
@@ -234,13 +246,12 @@ public class BlockIndexStore {
* @param blocksList blocks loaded in thread
* @throws IndexBuilderException in case of any failure
*/
- private void fillLoadedBlocks(AbstractIndex[] loadedBlockArray,
- List<Future<AbstractIndex>> blocksList) throws IndexBuilderException {
- int blockCounter = 0;
+ private void fillLoadedBlocks(AbstractIndex[] loadedBlockArray, List<BlockInfo> blockInfos)
+ throws IndexBuilderException {
for (int i = 0; i < loadedBlockArray.length; i++) {
if (null == loadedBlockArray[i]) {
try {
- loadedBlockArray[i] = blocksList.get(blockCounter++).get();
+ loadedBlockArray[i] = mapOfBlockInfoToFuture.get(blockInfos.get(i)).get();
} catch (InterruptedException | ExecutionException e) {
throw new IndexBuilderException(e);
}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/03066142/core/src/main/java/org/apache/carbondata/core/carbon/datastore/block/BlockInfo.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/carbon/datastore/block/BlockInfo.java b/core/src/main/java/org/apache/carbondata/core/carbon/datastore/block/BlockInfo.java
index 8092c7c..96879aa 100644
--- a/core/src/main/java/org/apache/carbondata/core/carbon/datastore/block/BlockInfo.java
+++ b/core/src/main/java/org/apache/carbondata/core/carbon/datastore/block/BlockInfo.java
@@ -91,7 +91,7 @@ public class BlockInfo {
if (info.getBlockOffset() != other.info.getBlockOffset()) {
return false;
}
- if (info.getBlockLength() != info.getBlockLength()) {
+ if (info.getBlockLength() != other.info.getBlockLength()) {
return false;
}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/03066142/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java b/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
index 997f412..e217d5d 100644
--- a/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
+++ b/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
@@ -882,7 +882,7 @@ public final class CarbonCommonConstants {
/**
* to enable blocklet distribution default value
*/
- public static String ENABLE_BLOCKLET_DISTRIBUTION_DEFAULTVALUE = "true";
+ public static String ENABLE_BLOCKLET_DISTRIBUTION_DEFAULTVALUE = "false";
/**
* This batch size is used to send rows from load step to another step in batches.
[4/5] incubator-carbondata git commit: In case while copying store to
hdfs, if stream didn't closed properly,
file doesn't get copied to HDFS So we need to propogate exception to executor
so that task can be launched again.
Posted by ra...@apache.org.
In case while copying store to hdfs, if stream didn't closed properly, file doesn't get copied to HDFS
So we need to propogate exception to executor so that task can be launched again.
Project: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/commit/bc3f954f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/tree/bc3f954f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/diff/bc3f954f
Branch: refs/heads/branch-0.2
Commit: bc3f954f7392ca092b80ccac6d16e76e9bd222aa
Parents: ad9a515
Author: Ashok <as...@gmail.com>
Authored: Thu Nov 3 07:41:07 2016 +0530
Committer: ravipesala <ra...@gmail.com>
Committed: Mon Nov 7 17:15:28 2016 +0530
----------------------------------------------------------------------
.../apache/carbondata/core/util/CarbonUtil.java | 21 ++++++++++++++------
.../store/writer/AbstractFactDataWriter.java | 3 ++-
2 files changed, 17 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/bc3f954f/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java b/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java
index 8aceecd..bb8e82b 100644
--- a/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java
+++ b/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java
@@ -117,18 +117,27 @@ public final class CarbonUtil {
// Added if to avoid NullPointerException in case one stream is being passed as null
if (null != streams) {
for (Closeable stream : streams) {
- if (null != stream) {
- try {
- stream.close();
- } catch (IOException e) {
- LOGGER.error("Error while closing stream" + stream);
- }
+ try {
+ closeStream(stream);
+ } catch (IOException e) {
+ LOGGER.error("Error while closing stream:" + e);
}
}
}
}
/**
+ * This method closes stream
+ *
+ * @param stream
+ * @throws IOException
+ */
+ public static void closeStream(Closeable stream) throws IOException {
+ if (null != stream) {
+ stream.close();
+ }
+ }
+ /**
* @param baseStorePath
* @return
*/
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/bc3f954f/processing/src/main/java/org/apache/carbondata/processing/store/writer/AbstractFactDataWriter.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/store/writer/AbstractFactDataWriter.java b/processing/src/main/java/org/apache/carbondata/processing/store/writer/AbstractFactDataWriter.java
index f56a5d2..16bd771 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/store/writer/AbstractFactDataWriter.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/store/writer/AbstractFactDataWriter.java
@@ -632,7 +632,8 @@ public abstract class AbstractFactDataWriter<T> implements CarbonFactDataWriter<
.getDataInputStream(localFilePath, FileFactory.getFileType(localFilePath), bufferSize);
IOUtils.copyBytes(dataInputStream, dataOutputStream, bufferSize);
} finally {
- CarbonUtil.closeStreams(dataInputStream, dataOutputStream);
+ CarbonUtil.closeStream(dataInputStream);
+ CarbonUtil.closeStream(dataOutputStream);
}
}
[3/5] incubator-carbondata git commit: Pass block size to model
during compaction.
Posted by ra...@apache.org.
Pass block size to model during compaction.
Project: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/commit/ad9a5155
Tree: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/tree/ad9a5155
Diff: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/diff/ad9a5155
Branch: refs/heads/branch-0.2
Commit: ad9a5155e1734177ddbad1d5265b3db887c0404a
Parents: 91d94e7
Author: Ashok <as...@gmail.com>
Authored: Thu Nov 3 07:35:37 2016 +0530
Committer: ravipesala <ra...@gmail.com>
Committed: Mon Nov 7 17:12:36 2016 +0530
----------------------------------------------------------------------
.../carbondata/integration/spark/merger/RowResultMerger.java | 5 +++--
1 file changed, 3 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/ad9a5155/integration/spark/src/main/java/org/apache/carbondata/integration/spark/merger/RowResultMerger.java
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/java/org/apache/carbondata/integration/spark/merger/RowResultMerger.java b/integration/spark/src/main/java/org/apache/carbondata/integration/spark/merger/RowResultMerger.java
index be2a81d..7c3d2a7 100644
--- a/integration/spark/src/main/java/org/apache/carbondata/integration/spark/merger/RowResultMerger.java
+++ b/integration/spark/src/main/java/org/apache/carbondata/integration/spark/merger/RowResultMerger.java
@@ -102,7 +102,8 @@ public class RowResultMerger {
this.tableName = tableName;
this.measureCount = segprop.getMeasures().size();
-
+ CarbonTable carbonTable = CarbonMetadata.getInstance()
+ .getCarbonTable(databaseName + CarbonCommonConstants.UNDERSCORE + tableName);
CarbonFactDataHandlerModel carbonFactDataHandlerModel =
getCarbonFactDataHandlerModel(loadModel);
carbonFactDataHandlerModel.setPrimitiveDimLens(segprop.getDimColumnsCardinality());
@@ -117,7 +118,7 @@ public class RowResultMerger {
carbonFactDataHandlerModel.setMdKeyIndex(measureCount);
}
carbonFactDataHandlerModel.setColCardinality(colCardinality);
-
+ carbonFactDataHandlerModel.setBlockSizeInMB(carbonTable.getBlockSizeInMB());
dataHandler = new CarbonFactDataHandlerColumnar(carbonFactDataHandlerModel);
tupleConvertor = new TupleConversionAdapter(segProp);
[5/5] incubator-carbondata git commit: Drop Database Cascade command
to be supported in Carbon.
Posted by ra...@apache.org.
Drop Database Cascade command to be supported in Carbon.
Project: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/commit/e0261831
Tree: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/tree/e0261831
Diff: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/diff/e0261831
Branch: refs/heads/branch-0.2
Commit: e0261831db61e6631cb0f38ae7e1761f0278a3b7
Parents: bc3f954
Author: nareshpr <pr...@gmail.com>
Authored: Thu Nov 3 14:53:03 2016 +0530
Committer: ravipesala <ra...@gmail.com>
Committed: Mon Nov 7 17:17:51 2016 +0530
----------------------------------------------------------------------
.../spark/sql/CarbonCatalystOperators.scala | 15 ++++++
.../org/apache/spark/sql/CarbonSqlParser.scala | 46 +++++++++++++---
.../spark/sql/hive/CarbonMetastoreCatalog.scala | 16 +++++-
.../spark/sql/hive/CarbonStrategies.scala | 9 ++++
.../execution/command/CarbonHiveCommands.scala | 55 ++++++++++++++++++++
.../deleteTable/TestDeleteTableNewDDL.scala | 19 ++++---
6 files changed, 145 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/e0261831/integration/spark/src/main/scala/org/apache/spark/sql/CarbonCatalystOperators.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/CarbonCatalystOperators.scala b/integration/spark/src/main/scala/org/apache/spark/sql/CarbonCatalystOperators.scala
index 3729d3c..cece5e6 100644
--- a/integration/spark/src/main/scala/org/apache/spark/sql/CarbonCatalystOperators.scala
+++ b/integration/spark/src/main/scala/org/apache/spark/sql/CarbonCatalystOperators.scala
@@ -90,3 +90,18 @@ abstract class CarbonProfile(attributes: Seq[Attribute]) extends Serializable {
case class IncludeProfile(attributes: Seq[Attribute]) extends CarbonProfile(attributes)
case class ExcludeProfile(attributes: Seq[Attribute]) extends CarbonProfile(attributes)
+
+case class CreateDatabase(dbName: String, sql: String) extends LogicalPlan with Command {
+ override def children: Seq[LogicalPlan] = Seq.empty
+ override def output: Seq[AttributeReference] = {
+ Seq()
+ }
+}
+
+case class DropDatabase(dbName: String, isCascade: Boolean, sql: String)
+ extends LogicalPlan with Command {
+ override def children: Seq[LogicalPlan] = Seq.empty
+ override def output: Seq[AttributeReference] = {
+ Seq()
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/e0261831/integration/spark/src/main/scala/org/apache/spark/sql/CarbonSqlParser.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/CarbonSqlParser.scala b/integration/spark/src/main/scala/org/apache/spark/sql/CarbonSqlParser.scala
index da14b3e..c287617 100644
--- a/integration/spark/src/main/scala/org/apache/spark/sql/CarbonSqlParser.scala
+++ b/integration/spark/src/main/scala/org/apache/spark/sql/CarbonSqlParser.scala
@@ -199,14 +199,50 @@ class CarbonSqlParser()
override protected lazy val start: Parser[LogicalPlan] = explainPlan | startCommand
- protected lazy val startCommand: Parser[LogicalPlan] =
- dropDatabaseCascade | loadManagement | describeTable | showLoads | alterTable | createTable
+ protected lazy val startCommand: Parser[LogicalPlan] = createDatabase | dropDatabase |
+ loadManagement | describeTable | showLoads | alterTable | createTable
protected lazy val loadManagement: Parser[LogicalPlan] = deleteLoadsByID | deleteLoadsByLoadDate |
cleanFiles | loadDataNew
protected val escapedIdentifier = "`([^`]+)`".r
+ protected lazy val createDatabase: Parser[LogicalPlan] =
+ CREATE ~> (DATABASE | SCHEMA) ~> restInput ^^ {
+ case statement =>
+ val createDbSql = "CREATE DATABASE " + statement
+ var dbName = ""
+ // Get Ast node for create db command
+ val node = HiveQlWrapper.getAst(createDbSql)
+ node match {
+ // get dbname
+ case Token("TOK_CREATEDATABASE", children) =>
+ dbName = BaseSemanticAnalyzer.unescapeIdentifier(children(0).getText)
+ }
+ CreateDatabase(dbName, createDbSql)
+ }
+
+ protected lazy val dropDatabase: Parser[LogicalPlan] =
+ DROP ~> (DATABASE | SCHEMA) ~> restInput ^^ {
+ case statement =>
+ val dropDbSql = "DROP DATABASE " + statement
+ var dbName = ""
+ var isCascade = false
+ // Get Ast node for drop db command
+ val node = HiveQlWrapper.getAst(dropDbSql)
+ node match {
+ case Token("TOK_DROPDATABASE", children) =>
+ dbName = BaseSemanticAnalyzer.unescapeIdentifier(children(0).getText)
+ // check whether cascade drop db
+ children.collect {
+ case t@Token("TOK_CASCADE", _) =>
+ isCascade = true
+ case _ => // Unsupport features
+ }
+ }
+ DropDatabase(dbName, isCascade, dropDbSql)
+ }
+
private def reorderDimensions(dims: Seq[Field]): Seq[Field] = {
var complexDimensions: Seq[Field] = Seq()
var dimensions: Seq[Field] = Seq()
@@ -1228,10 +1264,4 @@ class CarbonSqlParser()
case _ => ExplainCommand(OneRowRelation)
}
}
-
- protected lazy val dropDatabaseCascade: Parser[LogicalPlan] =
- DROP ~> (DATABASE|SCHEMA) ~> opt(IF ~> EXISTS) ~> ident ~> CASCADE <~ opt(";") ^^ {
- case cascade => throw new MalformedCarbonCommandException(
- "Unsupported cascade operation in drop database/schema command")
- }
}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/e0261831/integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonMetastoreCatalog.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonMetastoreCatalog.scala b/integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonMetastoreCatalog.scala
index 7b3fea9..75dc235 100644
--- a/integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonMetastoreCatalog.scala
+++ b/integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonMetastoreCatalog.scala
@@ -575,6 +575,21 @@ class CarbonMetastoreCatalog(hiveContext: HiveContext, val storePath: String,
(databaseName, tableName, dataPath, schema, partitioner, tableCreationTime)
}
+ def createDatabaseDirectory(dbName: String) {
+ val databasePath = storePath + File.separator + dbName
+ val fileType = FileFactory.getFileType(databasePath)
+ FileFactory.mkdirs(databasePath, fileType)
+ }
+
+ def dropDatabaseDirectory(dbName: String) {
+ val databasePath = storePath + File.separator + dbName
+ val fileType = FileFactory.getFileType(databasePath)
+ if (FileFactory.isFileExist(databasePath, fileType)) {
+ val dbPath = FileFactory.getCarbonFile(databasePath, fileType)
+ CarbonUtil.deleteFoldersAndFiles(dbPath)
+ }
+ }
+
}
@@ -653,5 +668,4 @@ object CarbonMetastoreTypes extends RegexParsers {
case TimestampType => "timestamp"
}
}
-
}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/e0261831/integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonStrategies.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonStrategies.scala b/integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonStrategies.scala
index 0b4d020..03ddeff 100644
--- a/integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonStrategies.scala
+++ b/integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonStrategies.scala
@@ -32,6 +32,7 @@ import org.apache.spark.sql.execution.{ExecutedCommand, Filter, Project, SparkPl
import org.apache.spark.sql.execution.command._
import org.apache.spark.sql.execution.datasources.{DescribeCommand => LogicalDescribeCommand, LogicalRelation}
import org.apache.spark.sql.hive.execution.{DropTable, HiveNativeCommand}
+import org.apache.spark.sql.hive.execution.command._
import org.apache.spark.sql.optimizer.{CarbonAliasDecoderRelation, CarbonDecoderRelation}
import org.apache.spark.sql.types.IntegerType
@@ -251,6 +252,14 @@ class CarbonStrategies(sqlContext: SQLContext) extends QueryPlanner[SparkPlan] {
} else {
ExecutedCommand(HiveNativeCommand(altertablemodel.alterSql)) :: Nil
}
+ case CreateDatabase(dbName, sql) =>
+ ExecutedCommand(CreateDatabaseCommand(dbName, HiveNativeCommand(sql))) :: Nil
+ case DropDatabase(dbName, isCascade, sql) =>
+ if (isCascade) {
+ ExecutedCommand(DropDatabaseCascadeCommand(dbName, HiveNativeCommand(sql))) :: Nil
+ } else {
+ ExecutedCommand(DropDatabaseCommand(dbName, HiveNativeCommand(sql))) :: Nil
+ }
case d: HiveNativeCommand =>
try {
val resolvedTable = sqlContext.executePlan(CarbonHiveSyntax.parse(d.sql)).optimizedPlan
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/e0261831/integration/spark/src/main/scala/org/apache/spark/sql/hive/execution/command/CarbonHiveCommands.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/hive/execution/command/CarbonHiveCommands.scala b/integration/spark/src/main/scala/org/apache/spark/sql/hive/execution/command/CarbonHiveCommands.scala
new file mode 100644
index 0000000..e81e0b5
--- /dev/null
+++ b/integration/spark/src/main/scala/org/apache/spark/sql/hive/execution/command/CarbonHiveCommands.scala
@@ -0,0 +1,55 @@
+/*
+ * 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.hive.execution.command
+
+import org.apache.spark.sql._
+import org.apache.spark.sql.execution.RunnableCommand
+import org.apache.spark.sql.execution.command.DropTableCommand
+import org.apache.spark.sql.hive.execution.HiveNativeCommand
+
+private[hive] case class CreateDatabaseCommand(dbName: String,
+ command: HiveNativeCommand) extends RunnableCommand {
+ def run(sqlContext: SQLContext): Seq[Row] = {
+ val rows = command.run(sqlContext)
+ CarbonEnv.getInstance(sqlContext).carbonCatalog.createDatabaseDirectory(dbName)
+ rows
+ }
+}
+
+private[hive] case class DropDatabaseCommand(dbName: String,
+ command: HiveNativeCommand) extends RunnableCommand {
+ def run(sqlContext: SQLContext): Seq[Row] = {
+ val rows = command.run(sqlContext)
+ CarbonEnv.getInstance(sqlContext).carbonCatalog.dropDatabaseDirectory(dbName)
+ rows
+ }
+}
+
+private[hive] case class DropDatabaseCascadeCommand(dbName: String,
+ command: HiveNativeCommand) extends RunnableCommand {
+ def run(sqlContext: SQLContext): Seq[Row] = {
+ val tablesInDB = CarbonEnv.getInstance(sqlContext).carbonCatalog
+ .getTables(Some(dbName))(sqlContext).map(x => x._1)
+ val rows = command.run(sqlContext)
+ tablesInDB.foreach{tableName =>
+ DropTableCommand(true, Some(dbName), tableName).run(sqlContext)
+ }
+ CarbonEnv.getInstance(sqlContext).carbonCatalog.dropDatabaseDirectory(dbName)
+ rows
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/e0261831/integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/deleteTable/TestDeleteTableNewDDL.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/deleteTable/TestDeleteTableNewDDL.scala b/integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/deleteTable/TestDeleteTableNewDDL.scala
index adec571..cdfe2e6 100644
--- a/integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/deleteTable/TestDeleteTableNewDDL.scala
+++ b/integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/deleteTable/TestDeleteTableNewDDL.scala
@@ -50,20 +50,27 @@ class TestDeleteTableNewDDL extends QueryTest with BeforeAndAfterAll {
// normal deletion case
test("drop table Test with new DDL") {
sql("drop table table1")
-
}
test("test drop database cascade command") {
sql("create database testdb")
+ sql("use testdb")
+ sql("CREATE TABLE IF NOT EXISTS testtable(empno Int, empname string, utilization Int,salary Int)"
+ + " STORED BY 'org.apache.carbondata.format' ")
+ try {
+ sql("drop database testdb")
+ assert(false)
+ } catch {
+ case e : Exception =>
+ }
+ sql("drop database testdb cascade")
try {
- sql("drop database testdb cascade")
+ sql("use testdb")
assert(false)
} catch {
- case e : MalformedCarbonCommandException => {
- assert(e.getMessage.equals("Unsupported cascade operation in drop database/schema command"))
- }
+ case e : Exception =>
}
- sql("drop database testdb")
+ sql("use default")
}
// deletion case with if exists
[2/5] incubator-carbondata git commit: When data load and drop is
executed concurrently, then even though drop didn't get lock, its table files.
Posted by ra...@apache.org.
When data load and drop is executed concurrently, then even though drop didn't get lock, its table files.
Project: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/commit/91d94e7a
Tree: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/tree/91d94e7a
Diff: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/diff/91d94e7a
Branch: refs/heads/branch-0.2
Commit: 91d94e7a203d5521d40866c2fc1ba1355b1e7577
Parents: 0306614
Author: Ashok <as...@gmail.com>
Authored: Thu Nov 3 07:37:59 2016 +0530
Committer: ravipesala <ra...@gmail.com>
Committed: Mon Nov 7 17:12:15 2016 +0530
----------------------------------------------------------------------
.../apache/spark/sql/execution/command/carbonTableSchema.scala | 6 ++++--
1 file changed, 4 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/91d94e7a/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala b/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala
index 5599ae0..0e87eb9 100644
--- a/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala
+++ b/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala
@@ -1258,8 +1258,10 @@ private[sql] case class DropTableCommand(ifExistsSet: Boolean, databaseNameOp: O
val carbonLock = CarbonLockFactory
.getCarbonLockObj(carbonTableIdentifier, LockUsage.DROP_TABLE_LOCK)
val storePath = CarbonEnv.getInstance(sqlContext).carbonCatalog.storePath
+ var isLocked = false
try {
- if (carbonLock.lockWithRetries()) {
+ isLocked = carbonLock.lockWithRetries()
+ if (isLocked) {
logInfo("Successfully able to get the lock for drop.")
}
else {
@@ -1271,7 +1273,7 @@ private[sql] case class DropTableCommand(ifExistsSet: Boolean, databaseNameOp: O
LOGGER.audit(s"Deleted table [$tableName] under database [$dbName]")
}
finally {
- if (carbonLock != null) {
+ if (carbonLock != null && isLocked) {
if (carbonLock.unlock()) {
logInfo("Table MetaData Unlocked Successfully after dropping the table")
// deleting any remaining files.