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 2017/11/14 17:50:36 UTC
[14/18] carbondata git commit: [CARBONDATA-1573] [Integration]
Support Database Location Configuration while Creating Database/ Support
Creation of carbon Table in the database location
http://git-wip-us.apache.org/repos/asf/carbondata/blob/1155d4d8/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/DataMapWriterSuite.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/DataMapWriterSuite.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/DataMapWriterSuite.scala
index 348d9a7..9545f0f 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/DataMapWriterSuite.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/DataMapWriterSuite.scala
@@ -62,7 +62,6 @@ class C2DataMapFactory() extends DataMapFactory {
}
class DataMapWriterSuite extends QueryTest with BeforeAndAfterAll {
-
def buildTestData(numRows: Int): DataFrame = {
import sqlContext.implicits._
sqlContext.sparkContext.parallelize(1 to numRows)
@@ -81,72 +80,74 @@ class DataMapWriterSuite extends QueryTest with BeforeAndAfterAll {
test("test write datamap 2 pages") {
// register datamap writer
- DataMapStoreManager.getInstance().createAndRegisterDataMap(
- AbsoluteTableIdentifier.from(storeLocation, "default", "carbon1"),
- classOf[C2DataMapFactory].getName,
- "test")
-
- val df = buildTestData(33000)
-
- // save dataframe to carbon file
- df.write
- .format("carbondata")
- .option("tableName", "carbon1")
- .mode(SaveMode.Overwrite)
- .save()
-
- assert(DataMapWriterSuite.callbackSeq.head.contains("block start"))
- assert(DataMapWriterSuite.callbackSeq.last.contains("block end"))
- assert(DataMapWriterSuite.callbackSeq.slice(1, DataMapWriterSuite.callbackSeq.length - 1) == Seq(
- "blocklet start 0",
- "add page data: blocklet 0, page 0",
- "add page data: blocklet 0, page 1",
- "blocklet end: 0"
- ))
- DataMapWriterSuite.callbackSeq = Seq()
+ DataMapStoreManager.getInstance().createAndRegisterDataMap(
+ AbsoluteTableIdentifier.from(storeLocation + "/carbon1", "default", "carbon1"),
+ classOf[C2DataMapFactory].getName,
+ "test")
+
+ val df = buildTestData(33000)
+
+ // save dataframe to carbon file
+ df.write
+ .format("carbondata")
+ .option("tableName", "carbon1")
+ .mode(SaveMode.Overwrite)
+ .save()
+
+ assert(DataMapWriterSuite.callbackSeq.head.contains("block start"))
+ assert(DataMapWriterSuite.callbackSeq.last.contains("block end"))
+ assert(
+ DataMapWriterSuite.callbackSeq.slice(1, DataMapWriterSuite.callbackSeq.length - 1) == Seq(
+ "blocklet start 0",
+ "add page data: blocklet 0, page 0",
+ "add page data: blocklet 0, page 1",
+ "blocklet end: 0"
+ ))
+ DataMapWriterSuite.callbackSeq = Seq()
}
test("test write datamap 2 blocklet") {
// register datamap writer
- DataMapStoreManager.getInstance().createAndRegisterDataMap(
- AbsoluteTableIdentifier.from(storeLocation, "default", "carbon2"),
- classOf[C2DataMapFactory].getName,
- "test")
+ DataMapStoreManager.getInstance().createAndRegisterDataMap(
+ AbsoluteTableIdentifier.from(storeLocation + "/carbon2", "default", "carbon2"),
+ classOf[C2DataMapFactory].getName,
+ "test")
- CarbonProperties.getInstance()
- .addProperty("carbon.blockletgroup.size.in.mb", "1")
+ CarbonProperties.getInstance()
+ .addProperty("carbon.blockletgroup.size.in.mb", "1")
CarbonProperties.getInstance()
.addProperty("carbon.number.of.cores.while.loading",
CarbonCommonConstants.NUM_CORES_DEFAULT_VAL)
- val df = buildTestData(300000)
-
- // save dataframe to carbon file
- df.write
- .format("carbondata")
- .option("tableName", "carbon2")
- .mode(SaveMode.Overwrite)
- .save()
-
- assert(DataMapWriterSuite.callbackSeq.head.contains("block start"))
- assert(DataMapWriterSuite.callbackSeq.last.contains("block end"))
- assert(DataMapWriterSuite.callbackSeq.slice(1, DataMapWriterSuite.callbackSeq.length - 1) == Seq(
- "blocklet start 0",
- "add page data: blocklet 0, page 0",
- "add page data: blocklet 0, page 1",
- "add page data: blocklet 0, page 2",
- "add page data: blocklet 0, page 3",
- "add page data: blocklet 0, page 4",
- "add page data: blocklet 0, page 5",
- "add page data: blocklet 0, page 6",
- "add page data: blocklet 0, page 7",
- "blocklet end: 0",
- "blocklet start 1",
- "add page data: blocklet 1, page 0",
- "add page data: blocklet 1, page 1",
- "blocklet end: 1"
- ))
- DataMapWriterSuite.callbackSeq = Seq()
+ val df = buildTestData(300000)
+
+ // save dataframe to carbon file
+ df.write
+ .format("carbondata")
+ .option("tableName", "carbon2")
+ .mode(SaveMode.Overwrite)
+ .save()
+
+ assert(DataMapWriterSuite.callbackSeq.head.contains("block start"))
+ assert(DataMapWriterSuite.callbackSeq.last.contains("block end"))
+ assert(
+ DataMapWriterSuite.callbackSeq.slice(1, DataMapWriterSuite.callbackSeq.length - 1) == Seq(
+ "blocklet start 0",
+ "add page data: blocklet 0, page 0",
+ "add page data: blocklet 0, page 1",
+ "add page data: blocklet 0, page 2",
+ "add page data: blocklet 0, page 3",
+ "add page data: blocklet 0, page 4",
+ "add page data: blocklet 0, page 5",
+ "add page data: blocklet 0, page 6",
+ "add page data: blocklet 0, page 7",
+ "blocklet end: 0",
+ "blocklet start 1",
+ "add page data: blocklet 1, page 0",
+ "add page data: blocklet 1, page 1",
+ "blocklet end: 1"
+ ))
+ DataMapWriterSuite.callbackSeq = Seq()
}
override def afterAll {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/1155d4d8/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/dataretention/DataRetentionTestCase.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/dataretention/DataRetentionTestCase.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/dataretention/DataRetentionTestCase.scala
index a16c5bf..2c6178d 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/dataretention/DataRetentionTestCase.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/dataretention/DataRetentionTestCase.scala
@@ -24,8 +24,9 @@ import org.apache.carbondata.core.locks.{CarbonLockFactory, ICarbonLock, LockUsa
import org.apache.commons.lang3.time.DateUtils
import org.apache.spark.sql.Row
import org.scalatest.BeforeAndAfterAll
-import org.apache.carbondata.core.util.path.CarbonStorePath
-import org.apache.carbondata.core.metadata.{AbsoluteTableIdentifier, CarbonTableIdentifier}
+
+import org.apache.carbondata.core.util.path.{CarbonStorePath, CarbonTablePath}
+import org.apache.carbondata.core.metadata.{AbsoluteTableIdentifier, CarbonMetadata, CarbonTableIdentifier}
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.util.CarbonProperties
import org.apache.carbondata.spark.exception.MalformedCarbonCommandException
@@ -36,28 +37,16 @@ import org.apache.spark.sql.test.util.QueryTest
*/
class DataRetentionTestCase extends QueryTest with BeforeAndAfterAll {
- val absoluteTableIdentifierForLock: AbsoluteTableIdentifier = new
- AbsoluteTableIdentifier(storeLocation,
- new CarbonTableIdentifier(CarbonCommonConstants.DATABASE_DEFAULT_NAME, "retentionlock", "200"))
- val absoluteTableIdentifierForRetention: AbsoluteTableIdentifier = new
- AbsoluteTableIdentifier(storeLocation,
- new CarbonTableIdentifier(
- CarbonCommonConstants.DATABASE_DEFAULT_NAME, "DataRetentionTable".toLowerCase(), "300"))
- val carbonTablePath = CarbonStorePath
- .getCarbonTablePath(absoluteTableIdentifierForRetention.getStorePath,
- absoluteTableIdentifierForRetention.getCarbonTableIdentifier).getMetadataDirectoryPath
-
+ var absoluteTableIdentifierForLock: AbsoluteTableIdentifier = null
+ var absoluteTableIdentifierForRetention: AbsoluteTableIdentifier = null
+ var carbonTablePath : String = null
var carbonDateFormat = new SimpleDateFormat(CarbonCommonConstants.CARBON_TIMESTAMP)
var defaultDateFormat = new SimpleDateFormat(CarbonCommonConstants
.CARBON_TIMESTAMP_DEFAULT_FORMAT)
- val carbonTableStatusLock: ICarbonLock = CarbonLockFactory
- .getCarbonLockObj(absoluteTableIdentifierForLock.getCarbonTableIdentifier, LockUsage.TABLE_STATUS_LOCK)
- val carbonDeleteSegmentLock: ICarbonLock = CarbonLockFactory
- .getCarbonLockObj(absoluteTableIdentifierForLock.getCarbonTableIdentifier, LockUsage.DELETE_SEGMENT_LOCK)
- val carbonCleanFilesLock: ICarbonLock = CarbonLockFactory
- .getCarbonLockObj(absoluteTableIdentifierForLock.getCarbonTableIdentifier, LockUsage.CLEAN_FILES_LOCK)
- val carbonMetadataLock: ICarbonLock = CarbonLockFactory
- .getCarbonLockObj(absoluteTableIdentifierForLock.getCarbonTableIdentifier, LockUsage.METADATA_LOCK)
+ var carbonTableStatusLock: ICarbonLock = null
+ var carbonDeleteSegmentLock: ICarbonLock = null
+ var carbonCleanFilesLock: ICarbonLock = null
+ var carbonMetadataLock: ICarbonLock = null
override def beforeAll {
sql("drop table if exists DataRetentionTable")
@@ -78,7 +67,20 @@ class DataRetentionTestCase extends QueryTest with BeforeAndAfterAll {
"phonetype String, serialname String, salary int) stored by 'org.apache.carbondata.format'"
)
-
+ val carbonTable = CarbonMetadata.getInstance().getCarbonTable(CarbonCommonConstants.DATABASE_DEFAULT_NAME+"_"+ "retentionlock")
+ absoluteTableIdentifierForLock = carbonTable.getAbsoluteTableIdentifier
+ val carbonTable2 = CarbonMetadata.getInstance().getCarbonTable(CarbonCommonConstants.DATABASE_DEFAULT_NAME+"_"+ "dataRetentionTable")
+ absoluteTableIdentifierForRetention = carbonTable2.getAbsoluteTableIdentifier
+ carbonTablePath = CarbonStorePath
+ .getCarbonTablePath(absoluteTableIdentifierForRetention).getMetadataDirectoryPath
+ carbonTableStatusLock = CarbonLockFactory
+ .getCarbonLockObj(absoluteTableIdentifierForLock, LockUsage.TABLE_STATUS_LOCK)
+ carbonDeleteSegmentLock= CarbonLockFactory
+ .getCarbonLockObj(absoluteTableIdentifierForLock, LockUsage.DELETE_SEGMENT_LOCK)
+ carbonCleanFilesLock = CarbonLockFactory
+ .getCarbonLockObj(absoluteTableIdentifierForLock, LockUsage.CLEAN_FILES_LOCK)
+ carbonMetadataLock = CarbonLockFactory
+ .getCarbonLockObj(absoluteTableIdentifierForLock, LockUsage.METADATA_LOCK)
sql(
s"LOAD DATA LOCAL INPATH '$resourcesPath/dataretention1.csv' INTO TABLE retentionlock " +
"OPTIONS('DELIMITER' = ',')")
http://git-wip-us.apache.org/repos/asf/carbondata/blob/1155d4d8/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/dblocation/DBLocationCarbonTableTestCase.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/dblocation/DBLocationCarbonTableTestCase.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/dblocation/DBLocationCarbonTableTestCase.scala
new file mode 100644
index 0000000..6f3191d
--- /dev/null
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/dblocation/DBLocationCarbonTableTestCase.scala
@@ -0,0 +1,223 @@
+/*
+ * 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.dblocation
+
+import org.apache.carbondata.core.constants.CarbonCommonConstants
+import org.apache.carbondata.core.datastore.impl.FileFactory
+import org.apache.carbondata.core.util.{CarbonProperties, CarbonUtil}
+import org.apache.spark.sql.{AnalysisException, Row}
+import org.apache.spark.sql.test.util.QueryTest
+import org.scalatest.BeforeAndAfterAll
+
+/**
+ *
+ */
+class DBLocationCarbonTableTestCase extends QueryTest with BeforeAndAfterAll {
+
+ def getMdtFileAndType() = {
+ // if mdt file path is configured then take configured path else take default path
+ val configuredMdtPath = CarbonProperties.getInstance()
+ .getProperty(CarbonCommonConstants.CARBON_UPDATE_SYNC_FOLDER,
+ CarbonCommonConstants.CARBON_UPDATE_SYNC_FOLDER_DEFAULT).trim
+ var timestampFile = configuredMdtPath + "/" + CarbonCommonConstants.SCHEMAS_MODIFIED_TIME_FILE
+ timestampFile = CarbonUtil.checkAndAppendFileSystemURIScheme(timestampFile)
+ val timestampFileType = FileFactory.getFileType(timestampFile)
+ (timestampFile, timestampFileType)
+
+ }
+
+ override def beforeAll {
+ sql("drop database if exists carbon cascade")
+ }
+
+ test("create and drop database test") {
+ sql(s"create database carbon location '$dblocation'")
+ sql("drop database if exists carbon cascade")
+ }
+
+ test("create two databases at same table") {
+ sql(s"create database carbon location '$dblocation'")
+ try {
+ sql(s"create database carbon1 location '$dblocation'")
+ } catch {
+ case e: AnalysisException =>
+ assert(true)
+ }
+ }
+
+ test("create table and load data") {
+ sql("drop database if exists carbon cascade")
+ sql(s"create database carbon location '$dblocation'")
+ sql("use carbon")
+ sql("""create table carbon.carbontable (c1 string,c2 int,c3 string,c5 string) STORED BY 'org.apache.carbondata.format'""")
+ sql(s"""LOAD DATA LOCAL INPATH '$resourcesPath/dblocation/test.csv' INTO table carbon.carbontable""")
+ checkAnswer(sql("select count(*) from carbontable"), Row(5))
+ }
+
+ test("create table and insert data") {
+ sql("drop database if exists carbon cascade")
+ sql(s"create database carbon location '$dblocation'")
+ sql("use carbon")
+ sql("""create table carbon.carbontable (c1 string,c2 int,c3 string,c5 string) STORED BY 'org.apache.carbondata.format'""")
+ sql("insert into carbontable select 'a',1,'aa','aaa'")
+ checkAnswer(sql("select count(*) from carbontable"), Row(1))
+ checkAnswer(sql("select c1 from carbontable"), Seq(Row("a")))
+ }
+
+ test("create table and 2 times data load") {
+ sql("drop database if exists carbon cascade")
+ sql(s"create database carbon location '$dblocation'")
+ sql("use carbon")
+ sql("""create table carbon.carbontable (c1 string,c2 int,c3 string,c5 string) STORED BY 'org.apache.carbondata.format'""")
+ sql("insert into carbontable select 'a',1,'aa','aaa'")
+ sql("insert into carbontable select 'b',1,'aa','aaa'")
+ checkAnswer(sql("select count(*) from carbontable"), Row(2))
+ checkAnswer(sql("select c1 from carbontable"), Seq(Row("a"), Row("b")))
+ }
+
+ //TODO fix this test case
+ test("Update operation on carbon table with insert into") {
+ sql("drop database if exists carbon cascade")
+ sql(s"create database carbon location '$dblocation'")
+ sql("use carbon")
+ sql("""create table carbon.carbontable (c1 string,c2 int,c3 string,c5 string) STORED BY 'org.apache.carbondata.format'""")
+ sql("insert into carbontable select 'a',1,'aa','aaa'")
+ sql("insert into carbontable select 'b',1,'bb','bbb'")
+ // update operation
+ sql("""update carbon.carbontable d set (d.c2) = (d.c2 + 1) where d.c1 = 'a'""").show()
+ sql("""update carbon.carbontable d set (d.c2) = (d.c2 + 1) where d.c1 = 'b'""").show()
+ checkAnswer(
+ sql("""select c1,c2,c3,c5 from carbon.carbontable"""),
+ Seq(Row("a",2,"aa","aaa"),Row("b",2,"bb","bbb"))
+ )
+ }
+
+
+ test("Update operation on carbon table") {
+ sql("drop database if exists carbon cascade")
+ sql(s"create database carbon location '$dblocation'")
+ sql("use carbon")
+ sql(
+ """
+ CREATE TABLE automerge(id int, name string, city string, age int)
+ STORED BY 'org.apache.carbondata.format'
+ """)
+ val testData = s"$resourcesPath/sample.csv"
+ sql(s"LOAD DATA LOCAL INPATH '$testData' into table automerge")
+ // update operation
+ sql("""update carbon.automerge d set (d.id) = (d.id + 1) where d.id > 2""").show()
+ checkAnswer(
+ sql("select count(*) from automerge"),
+ Seq(Row(6))
+ )
+ // sql("drop table carbontable")
+ }
+
+ test("Delete operation on carbon table") {
+ sql("drop database if exists carbon cascade")
+ sql(s"create database carbon location '$dblocation'")
+ sql("use carbon")
+ sql("""create table carbon.carbontable (c1 string,c2 int,c3 string,c5 string) STORED BY 'org.apache.carbondata.format'""")
+ sql("insert into carbontable select 'a',1,'aa','aaa'")
+ sql("insert into carbontable select 'b',1,'bb','bbb'")
+ // delete operation
+ sql("""delete from carbontable where c3 = 'aa'""").show
+ checkAnswer(
+ sql("""select c1,c2,c3,c5 from carbon.carbontable"""),
+ Seq(Row("b",1,"bb","bbb"))
+ )
+ sql("drop table carbontable")
+ }
+
+ test("Alter table add column test") {
+ sql("drop database if exists carbon cascade")
+ sql(s"create database carbon location '$dblocation'")
+ sql("use carbon")
+ sql("""create table carbon.carbontable (c1 string,c2 int,c3 string,c5 string) STORED BY 'org.apache.carbondata.format'""")
+ sql("insert into carbontable select 'a',1,'aa','aaa'")
+ sql("insert into carbontable select 'b',1,'bb','bbb'")
+ sql("Alter table carbontable add columns(c4 string) " +
+ "TBLPROPERTIES('DICTIONARY_EXCLUDE'='c4', 'DEFAULT.VALUE.c4'='def')")
+ checkAnswer(
+ sql("""select c1,c2,c3,c5,c4 from carbon.carbontable"""),
+ Seq(Row("a",1,"aa","aaa","def"), Row("b",1,"bb","bbb","def"))
+ )
+ sql("drop table carbontable")
+ }
+
+ test("Alter table change column datatype test") {
+ sql("drop database if exists carbon cascade")
+ sql(s"create database carbon location '$dblocation'")
+ sql("use carbon")
+ sql("""create table carbon.carbontable (c1 string,c2 int,c3 string,c5 string) STORED BY 'org.apache.carbondata.format'""")
+ sql("insert into carbontable select 'a',1,'aa','aaa'")
+ sql("insert into carbontable select 'b',1,'bb','bbb'")
+ sql("Alter table carbontable change c2 c2 long")
+ checkAnswer(
+ sql("""select c1,c2,c3,c5 from carbon.carbontable"""),
+ Seq(Row("a",1,"aa","aaa"), Row("b",1,"bb","bbb"))
+ )
+ sql("drop table carbontable")
+ }
+
+ test("Alter table drop column test") {
+ sql("drop database if exists carbon cascade")
+ sql(s"create database carbon location '$dblocation'")
+ sql("use carbon")
+ sql("""create table carbon.carbontable (c1 string,c2 int,c3 string,c5 string) STORED BY 'org.apache.carbondata.format'""")
+ sql("insert into carbontable select 'a',1,'aa','aaa'")
+ sql("insert into carbontable select 'b',1,'bb','bbb'")
+ sql("Alter table carbontable drop columns(c2)")
+ checkAnswer(
+ sql("""select * from carbon.carbontable"""),
+ Seq(Row("a","aa","aaa"), Row("b","bb","bbb"))
+ )
+ sql("drop table carbontable")
+ }
+
+ test("test mdt file path with configured paths") {
+ sql("drop database if exists carbon cascade")
+ sql(s"create database carbon location '$dblocation'")
+ sql("use carbon")
+ CarbonProperties.getInstance()
+ .addProperty(CarbonCommonConstants.CARBON_UPDATE_SYNC_FOLDER, "/tmp/carbondata1/carbondata2/")
+ val (timestampFile, timestampFileType) = getMdtFileAndType()
+ FileFactory.deleteFile(timestampFile, timestampFileType)
+ sql("""create table carbon.carbontable (c1 string,c2 int,c3 string,c5 string) STORED BY 'org.apache.carbondata.format'""")
+ sql("drop table carbontable")
+ // perform file check
+ assert(FileFactory.isFileExist(timestampFile, timestampFileType, true))
+
+ CarbonProperties.getInstance()
+ .addProperty(CarbonCommonConstants.CARBON_UPDATE_SYNC_FOLDER,
+ CarbonCommonConstants.CARBON_UPDATE_SYNC_FOLDER_DEFAULT)
+ val (timestampFile2, timestampFileType2) = getMdtFileAndType()
+ FileFactory.deleteFile(timestampFile2, timestampFileType2)
+ sql("""create table carbon.carbontable (c1 string,c2 int,c3 string,c5 string) STORED BY 'org.apache.carbondata.format'""")
+ sql("drop table carbontable")
+ // perform file check
+ assert(FileFactory.isFileExist(timestampFile, timestampFileType, true))
+ }
+
+ override def afterAll {
+ CarbonProperties.getInstance()
+ .addProperty(CarbonCommonConstants.CARBON_UPDATE_SYNC_FOLDER,
+ CarbonCommonConstants.CARBON_UPDATE_SYNC_FOLDER_DEFAULT)
+ sql("use default")
+ sql("drop database if exists carbon cascade")
+ }
+}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/1155d4d8/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/partition/TestDataLoadingForPartitionTable.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/partition/TestDataLoadingForPartitionTable.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/partition/TestDataLoadingForPartitionTable.scala
index 39777de..ed151bd 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/partition/TestDataLoadingForPartitionTable.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/partition/TestDataLoadingForPartitionTable.scala
@@ -61,8 +61,8 @@ class TestDataLoadingForPartitionTable extends QueryTest with BeforeAndAfterAll
def validateDataFiles(tableUniqueName: String, segmentId: String, partitions: Seq[Int]): Unit = {
val carbonTable = CarbonMetadata.getInstance().getCarbonTable(tableUniqueName)
- val tablePath = new CarbonTablePath(carbonTable.getStorePath, carbonTable.getDatabaseName,
- carbonTable.getFactTableName)
+ val tablePath = new CarbonTablePath(carbonTable.getCarbonTableIdentifier,
+ carbonTable.getTablePath)
val segmentDir = tablePath.getCarbonDataDirectoryPath("0", segmentId)
val carbonFile = FileFactory.getCarbonFile(segmentDir, FileFactory.getFileType(segmentDir))
val dataFiles = carbonFile.listFiles(new CarbonFileFilter() {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/1155d4d8/integration/spark-common/src/main/scala/org/apache/carbondata/api/CarbonStore.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/api/CarbonStore.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/api/CarbonStore.scala
index cf6a032..3d93a65 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/api/CarbonStore.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/api/CarbonStore.scala
@@ -105,21 +105,26 @@ object CarbonStore {
forceTableClean: Boolean): Unit = {
LOGGER.audit(s"The clean files request has been received for $dbName.$tableName")
var carbonCleanFilesLock: ICarbonLock = null
- val identifier = new CarbonTableIdentifier(dbName, tableName, "")
+ var absoluteTableIdentifier: AbsoluteTableIdentifier = null
+ if (forceTableClean) {
+ absoluteTableIdentifier = AbsoluteTableIdentifier.from(storePath, dbName, tableName)
+ } else {
+ absoluteTableIdentifier = carbonTable.getAbsoluteTableIdentifier
+ }
try {
val errorMsg = "Clean files request is failed for " +
s"$dbName.$tableName" +
". Not able to acquire the clean files lock due to another clean files " +
"operation is running in the background."
carbonCleanFilesLock =
- CarbonLockUtil.getLockObject(identifier, LockUsage.CLEAN_FILES_LOCK, errorMsg)
+ CarbonLockUtil.getLockObject(absoluteTableIdentifier, LockUsage.CLEAN_FILES_LOCK, errorMsg)
if (forceTableClean) {
val absIdent = AbsoluteTableIdentifier.from(storePath, dbName, tableName)
FileFactory.deleteAllCarbonFilesOfDir(
FileFactory.getCarbonFile(absIdent.getTablePath,
FileFactory.getFileType(absIdent.getTablePath)))
} else {
- DataManagementFunc.deleteLoadsAndUpdateMetadata(dbName, tableName, storePath,
+ DataManagementFunc.deleteLoadsAndUpdateMetadata(
isForceDeletion = true, carbonTable)
CarbonUpdateUtil.cleanUpDeltaFiles(carbonTable, true)
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/1155d4d8/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/AlterTableAddColumnRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/AlterTableAddColumnRDD.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/AlterTableAddColumnRDD.scala
index 2fbc7a3..c86f7d3 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/AlterTableAddColumnRDD.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/AlterTableAddColumnRDD.scala
@@ -23,7 +23,7 @@ import org.apache.spark.rdd.RDD
import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.datastore.impl.FileFactory
-import org.apache.carbondata.core.metadata.CarbonTableIdentifier
+import org.apache.carbondata.core.metadata.{AbsoluteTableIdentifier, CarbonTableIdentifier}
import org.apache.carbondata.core.metadata.encoder.Encoding
import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema
import org.apache.carbondata.core.statusmanager.SegmentStatus
@@ -51,8 +51,7 @@ class AddColumnPartition(rddId: Int, idx: Int, schema: ColumnSchema) extends Par
*/
class AlterTableAddColumnRDD[K, V](sc: SparkContext,
@transient newColumns: Seq[ColumnSchema],
- carbonTableIdentifier: CarbonTableIdentifier,
- carbonStorePath: String)
+ absoluteTableIdentifier: AbsoluteTableIdentifier)
extends CarbonRDD[(Int, SegmentStatus)](sc, Nil) {
val lockType: String = CarbonProperties.getInstance.getProperty(CarbonCommonConstants.LOCK_TYPE,
@@ -75,7 +74,7 @@ class AlterTableAddColumnRDD[K, V](sc: SparkContext,
if (columnSchema.hasEncoding(Encoding.DICTIONARY) &&
!columnSchema.hasEncoding(Encoding.DIRECT_DICTIONARY)) {
val carbonTablePath = CarbonStorePath
- .getCarbonTablePath(carbonStorePath, carbonTableIdentifier)
+ .getCarbonTablePath(absoluteTableIdentifier)
var rawData: String = null
if (null != columnSchema.getDefaultValue) {
rawData = new String(columnSchema.getDefaultValue,
@@ -91,8 +90,7 @@ class AlterTableAddColumnRDD[K, V](sc: SparkContext,
GlobalDictionaryUtil
.loadDefaultDictionaryValueForNewColumn(carbonTablePath,
columnSchema,
- carbonTableIdentifier,
- carbonStorePath,
+ absoluteTableIdentifier,
rawData)
}
} catch {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/1155d4d8/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/AlterTableDropColumnRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/AlterTableDropColumnRDD.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/AlterTableDropColumnRDD.scala
index 5b0889f..248f351 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/AlterTableDropColumnRDD.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/AlterTableDropColumnRDD.scala
@@ -18,12 +18,11 @@
package org.apache.carbondata.spark.rdd
import org.apache.spark.{Partition, SparkContext, TaskContext}
-import org.apache.spark.rdd.RDD
import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.cache.dictionary.ManageDictionaryAndBTree
import org.apache.carbondata.core.constants.CarbonCommonConstants
-import org.apache.carbondata.core.metadata.CarbonTableIdentifier
+import org.apache.carbondata.core.metadata.{AbsoluteTableIdentifier, CarbonTableIdentifier}
import org.apache.carbondata.core.metadata.encoder.Encoding
import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema
import org.apache.carbondata.core.statusmanager.SegmentStatus
@@ -48,8 +47,7 @@ class DropColumnPartition(rddId: Int, idx: Int, schema: ColumnSchema) extends Pa
*/
class AlterTableDropColumnRDD[K, V](sc: SparkContext,
@transient newColumns: Seq[ColumnSchema],
- carbonTableIdentifier: CarbonTableIdentifier,
- carbonStorePath: String)
+ carbonTableIdentifier: AbsoluteTableIdentifier)
extends CarbonRDD[(Int, SegmentStatus)](sc, Nil) {
override def getPartitions: Array[Partition] = {
@@ -68,7 +66,7 @@ class AlterTableDropColumnRDD[K, V](sc: SparkContext,
if (columnSchema.hasEncoding(Encoding.DICTIONARY) &&
!columnSchema.hasEncoding(Encoding.DIRECT_DICTIONARY)) {
ManageDictionaryAndBTree
- .deleteDictionaryFileAndCache(columnSchema, carbonTableIdentifier, carbonStorePath)
+ .deleteDictionaryFileAndCache(columnSchema, carbonTableIdentifier)
}
} catch {
case ex: Exception =>
http://git-wip-us.apache.org/repos/asf/carbondata/blob/1155d4d8/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonGlobalDictionaryRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonGlobalDictionaryRDD.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonGlobalDictionaryRDD.scala
index c1c7bf1..e7a4a1c 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonGlobalDictionaryRDD.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonGlobalDictionaryRDD.scala
@@ -38,7 +38,7 @@ import org.apache.carbondata.core.cache.dictionary.{Dictionary, DictionaryColumn
import org.apache.carbondata.core.constants.{CarbonCommonConstants, CarbonLoadOptionConstants}
import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.locks.{CarbonLockFactory, LockUsage}
-import org.apache.carbondata.core.metadata.{CarbonTableIdentifier, ColumnIdentifier}
+import org.apache.carbondata.core.metadata.{AbsoluteTableIdentifier, CarbonTableIdentifier, ColumnIdentifier}
import org.apache.carbondata.core.metadata.schema.table.column.CarbonDimension
import org.apache.carbondata.core.service.{CarbonCommonFactory, PathService}
import org.apache.carbondata.core.statusmanager.SegmentStatus
@@ -151,7 +151,7 @@ case class DataFormat(delimiters: Array[String],
/**
* a case class to package some attributes
*/
-case class DictionaryLoadModel(table: CarbonTableIdentifier,
+case class DictionaryLoadModel(table: AbsoluteTableIdentifier,
dimensions: Array[CarbonDimension],
hdfsLocation: String,
dictfolderPath: String,
@@ -280,8 +280,6 @@ class CarbonBlockDistinctValuesCombineRDD(
override def internalCompute(split: Partition,
context: TaskContext): Iterator[(Int, ColumnDistinctValues)] = {
val LOGGER = LogServiceFactory.getLogService(this.getClass.getName)
- CarbonProperties.getInstance().addProperty(CarbonCommonConstants.STORE_LOCATION,
- model.hdfsLocation)
CarbonTimeStatisticsFactory.getLoadStatisticsInstance.recordLoadCsvfilesToDfTime()
val distinctValuesList = new ArrayBuffer[(Int, mutable.HashSet[String])]
var rowCount = 0L
@@ -341,8 +339,6 @@ class CarbonGlobalDictionaryGenerateRDD(
override def internalCompute(split: Partition,
context: TaskContext): Iterator[(Int, SegmentStatus)] = {
val LOGGER = LogServiceFactory.getLogService(this.getClass.getName)
- CarbonProperties.getInstance().addProperty(CarbonCommonConstants.STORE_LOCATION,
- model.hdfsLocation)
var status = SegmentStatus.SUCCESS
val iter = new Iterator[(Int, SegmentStatus)] {
var dictionaryForDistinctValueLookUp: Dictionary = _
@@ -353,11 +349,11 @@ class CarbonGlobalDictionaryGenerateRDD(
model.table,
model.columnIdentifier(split.index),
model.columnIdentifier(split.index).getDataType,
- CarbonStorePath.getCarbonTablePath(model.hdfsLocation, model.table))
+ CarbonStorePath.getCarbonTablePath(model.table))
val pathService: PathService = CarbonCommonFactory.getPathService
val carbonTablePath: CarbonTablePath =
pathService
- .getCarbonTablePath(model.hdfsLocation, model.table, dictionaryColumnUniqueIdentifier)
+ .getCarbonTablePath(model.table, dictionaryColumnUniqueIdentifier)
if (StringUtils.isNotBlank(model.hdfsTempLocation)) {
CarbonProperties.getInstance.addProperty(CarbonCommonConstants.HDFS_TEMP_LOCATION,
model.hdfsTempLocation)
@@ -371,7 +367,7 @@ class CarbonGlobalDictionaryGenerateRDD(
model.zooKeeperUrl)
}
val dictLock = CarbonLockFactory
- .getCarbonLockObj(carbonTablePath.getRelativeDictionaryDirectory,
+ .getCarbonLockObj(model.table,
model.columnIdentifier(split.index).getColumnId + LockUsage.LOCK)
var isDictionaryLocked = false
// generate distinct value list
@@ -405,7 +401,6 @@ class CarbonGlobalDictionaryGenerateRDD(
dictionaryForDistinctValueLookUp = if (isDictFileExists) {
CarbonLoaderUtil.getDictionary(model.table,
model.columnIdentifier(split.index),
- model.hdfsLocation,
model.primDimensions(split.index).getDataType
)
} else {
@@ -415,9 +410,7 @@ class CarbonGlobalDictionaryGenerateRDD(
val t3 = System.currentTimeMillis()
val dictWriteTask = new DictionaryWriterTask(valuesBuffer,
dictionaryForDistinctValueLookUp,
- model.table,
dictionaryColumnUniqueIdentifier,
- model.hdfsLocation,
model.primDimensions(split.index).getColumnSchema,
isDictFileExists
)
@@ -427,10 +420,8 @@ class CarbonGlobalDictionaryGenerateRDD(
val t4 = System.currentTimeMillis()
// if new data came than rewrite sort index file
if (distinctValues.size() > 0) {
- val sortIndexWriteTask = new SortIndexWriterTask(model.table,
- dictionaryColumnUniqueIdentifier,
+ val sortIndexWriteTask = new SortIndexWriterTask(dictionaryColumnUniqueIdentifier,
model.primDimensions(split.index).getDataType,
- model.hdfsLocation,
dictionaryForDistinctValueLookUp,
distinctValues)
sortIndexWriteTask.execute()
http://git-wip-us.apache.org/repos/asf/carbondata/blob/1155d4d8/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonMergerRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonMergerRDD.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonMergerRDD.scala
index c6edc2a..2a7ca47 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonMergerRDD.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonMergerRDD.scala
@@ -167,7 +167,7 @@ class CarbonMergerRDD[K, V](
val dataFileMetadataSegMapping: java.util.Map[String, List[DataFileFooter]] =
CarbonCompactionUtil.createDataFileFooterMappingForSegments(tableBlockInfoList)
- carbonLoadModel.setStorePath(hdfsStoreLocation)
+ carbonLoadModel.setTablePath(hdfsStoreLocation)
// check for restructured block
// TODO: only in case of add and drop this variable should be true
val restructuredBlockExists: Boolean = CarbonCompactionUtil
http://git-wip-us.apache.org/repos/asf/carbondata/blob/1155d4d8/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanPartitionRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanPartitionRDD.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanPartitionRDD.scala
index 80cbab8..13a872f 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanPartitionRDD.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanPartitionRDD.scala
@@ -54,12 +54,12 @@ import org.apache.carbondata.processing.util.CarbonLoaderUtil
* This RDD is used in alter table partition statement to get data of target partitions,
* then repartition data according to new partitionInfo
* @param alterPartitionModel
- * @param carbonTableIdentifier
+ * @param absoluteTableIdentifier
* @param partitionIds the ids of target partition to be scanned
* @param bucketId
*/
class CarbonScanPartitionRDD(alterPartitionModel: AlterPartitionModel,
- carbonTableIdentifier: CarbonTableIdentifier,
+ absoluteTableIdentifier: AbsoluteTableIdentifier,
partitionIds: Seq[String],
bucketId: Int)
extends RDD[(AnyRef, Array[AnyRef])](alterPartitionModel.sqlContext.sparkContext, Nil) {
@@ -69,15 +69,14 @@ class CarbonScanPartitionRDD(alterPartitionModel: AlterPartitionModel,
val segmentId = alterPartitionModel.segmentId
val carbonLoadModel = alterPartitionModel.carbonLoadModel
val oldPartitionIdList = alterPartitionModel.oldPartitionIds
- val storePath = carbonLoadModel.getStorePath
- val identifier = new AbsoluteTableIdentifier(storePath, carbonTableIdentifier)
var storeLocation: String = null
var splitStatus: Boolean = false
var blockId: String = null
val carbonTable = carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable
val dimensions = carbonTable.getAllDimensions.asScala
val measures = carbonTable.getAllMeasures.asScala
- val partitionInfo = carbonTable.getPartitionInfo(carbonTableIdentifier.getTableName)
+ val partitionInfo = carbonTable
+ .getPartitionInfo(absoluteTableIdentifier.getCarbonTableIdentifier.getTableName)
val partitionColumn = partitionInfo.getColumnSchemaList().get(0)
val partitionDataType = partitionColumn.getDataType
val partitionColumnName = partitionColumn.getColumnName
@@ -94,7 +93,7 @@ class CarbonScanPartitionRDD(alterPartitionModel: AlterPartitionModel,
val parallelism = sparkContext.defaultParallelism
val jobConf = new JobConf(new Configuration)
val job = new Job(jobConf)
- val format = CarbonInputFormatUtil.createCarbonTableInputFormat(identifier,
+ val format = CarbonInputFormatUtil.createCarbonTableInputFormat(absoluteTableIdentifier,
partitionIds.toList.asJava, job)
job.getConfiguration.set("query.id", queryId)
@@ -114,7 +113,7 @@ class CarbonScanPartitionRDD(alterPartitionModel: AlterPartitionModel,
val splits = blocksPerTask.asScala.map(_.asInstanceOf[CarbonInputSplit])
if (blocksPerTask.size() != 0) {
val multiBlockSplit =
- new CarbonMultiBlockSplit(identifier, splits.asJava, Array(node))
+ new CarbonMultiBlockSplit(absoluteTableIdentifier, splits.asJava, Array(node))
val partition = new CarbonSparkPartition(id, partition_num, multiBlockSplit)
result.add(partition)
partition_num += 1
@@ -150,8 +149,8 @@ class CarbonScanPartitionRDD(alterPartitionModel: AlterPartitionModel,
sys.error("Exception occurred in query execution. Please check logs.")
}
}
- val segmentProperties = PartitionUtils.getSegmentProperties(identifier, segmentId,
- partitionIds.toList, oldPartitionIdList, partitionInfo)
+ val segmentProperties = PartitionUtils.getSegmentProperties(absoluteTableIdentifier,
+ segmentId, partitionIds.toList, oldPartitionIdList, partitionInfo)
val partColIdx = getPartitionColumnIndex(partitionColumnName, segmentProperties)
indexInitialise()
for (iterator <- result.asScala) {
@@ -202,8 +201,8 @@ class CarbonScanPartitionRDD(alterPartitionModel: AlterPartitionModel,
partitionValue = partitionValue.toString
}
} else { // normal dictionary
- val dict = CarbonLoaderUtil.getDictionary(carbonTableIdentifier,
- dimension.getColumnIdentifier, storePath, partitionDataType)
+ val dict = CarbonLoaderUtil.getDictionary(absoluteTableIdentifier,
+ dimension.getColumnIdentifier, partitionDataType)
if (partitionDataType == DataTypes.STRING) {
if (partitionType == PartitionType.RANGE) {
partitionValue = ByteUtil.
http://git-wip-us.apache.org/repos/asf/carbondata/blob/1155d4d8/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanRDD.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanRDD.scala
index 52a31a9..a84b040 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanRDD.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanRDD.scala
@@ -328,12 +328,21 @@ class CarbonScanRDD(
private def prepareInputFormatForDriver(conf: Configuration): CarbonTableInputFormat[Object] = {
CarbonTableInputFormat.setTableInfo(conf, tableInfo)
+ CarbonTableInputFormat.setDatabaseName(conf, tableInfo.getDatabaseName)
+ CarbonTableInputFormat
+ .setTableName(conf,
+ tableInfo.getOrCreateAbsoluteTableIdentifier().getCarbonTableIdentifier.getTableName)
createInputFormat(conf)
}
private def prepareInputFormatForExecutor(conf: Configuration): CarbonTableInputFormat[Object] = {
CarbonTableInputFormat.setCarbonReadSupport(conf, readSupport)
- CarbonTableInputFormat.setTableInfo(conf, getTableInfo)
+ val tableInfo1 = getTableInfo
+ CarbonTableInputFormat.setTableInfo(conf, tableInfo1)
+ CarbonTableInputFormat.setDatabaseName(conf, tableInfo1.getDatabaseName)
+ CarbonTableInputFormat
+ .setTableName(conf,
+ tableInfo1.getOrCreateAbsoluteTableIdentifier().getCarbonTableIdentifier.getTableName)
CarbonTableInputFormat.setDataTypeConverter(conf, new SparkDataTypeConverterImpl)
createInputFormat(conf)
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/1155d4d8/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/Compactor.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/Compactor.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/Compactor.scala
index adc71cf..a787af2 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/Compactor.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/Compactor.scala
@@ -38,13 +38,12 @@ object Compactor {
def triggerCompaction(compactionCallableModel: CompactionCallableModel): Unit = {
- val storeLocation = compactionCallableModel.storeLocation
val carbonTable = compactionCallableModel.carbonTable
val loadsToMerge = compactionCallableModel.loadsToMerge
val sc = compactionCallableModel.sqlContext
val carbonLoadModel = compactionCallableModel.carbonLoadModel
val compactionType = compactionCallableModel.compactionType
- val storePath = carbonLoadModel.getStorePath
+ val storePath = carbonLoadModel.getTablePath
val startTime = System.nanoTime()
val mergedLoadName = CarbonDataMergerUtil.getMergedLoadName(loadsToMerge)
var finalMergeStatus = false
@@ -53,8 +52,7 @@ object Compactor {
val validSegments: Array[String] = CarbonDataMergerUtil
.getValidSegments(loadsToMerge).split(',')
val mergeLoadStartTime = CarbonUpdateUtil.readCurrentTime()
- val carbonMergerMapping = CarbonMergerMapping(storeLocation,
- storePath,
+ val carbonMergerMapping = CarbonMergerMapping(storePath,
carbonTable.getMetaDataFilepath,
mergedLoadName,
databaseName,
@@ -65,7 +63,7 @@ object Compactor {
maxSegmentColCardinality = null,
maxSegmentColumnSchemaList = null
)
- carbonLoadModel.setStorePath(carbonMergerMapping.hdfsStoreLocation)
+ carbonLoadModel.setTablePath(carbonMergerMapping.hdfsStoreLocation)
carbonLoadModel.setLoadMetadataDetails(
SegmentStatusManager.readLoadMetadata(carbonTable.getMetaDataFilepath).toList.asJava)
// trigger event for compaction
http://git-wip-us.apache.org/repos/asf/carbondata/blob/1155d4d8/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/DataManagementFunc.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/DataManagementFunc.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/DataManagementFunc.scala
index 3a74dfa..762dcee 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/DataManagementFunc.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/DataManagementFunc.scala
@@ -76,8 +76,7 @@ object DataManagementFunc {
executor,
sqlContext,
compactionModel,
- carbonLoadModel,
- storeLocation
+ carbonLoadModel
)
try {
@@ -128,8 +127,7 @@ object DataManagementFunc {
executor: ExecutorService,
sqlContext: SQLContext,
compactionModel: CompactionModel,
- carbonLoadModel: CarbonLoadModel,
- storeLocation: String
+ carbonLoadModel: CarbonLoadModel
): Unit = {
loadsToMerge.asScala.foreach { seg =>
LOGGER.info("loads identified for merge is " + seg.getLoadName)
@@ -137,7 +135,6 @@ object DataManagementFunc {
val compactionCallableModel = CompactionCallableModel(
carbonLoadModel,
- storeLocation,
compactionModel.carbonTable,
loadsToMerge,
sqlContext,
@@ -175,25 +172,21 @@ object DataManagementFunc {
}
def deleteLoadsAndUpdateMetadata(
- dbName: String,
- tableName: String,
- storePath: String,
isForceDeletion: Boolean,
carbonTable: CarbonTable): Unit = {
if (isLoadDeletionRequired(carbonTable.getMetaDataFilepath)) {
val details = SegmentStatusManager.readLoadMetadata(carbonTable.getMetaDataFilepath)
+ val absoluteTableIdentifier = carbonTable.getAbsoluteTableIdentifier
val carbonTableStatusLock =
CarbonLockFactory.getCarbonLockObj(
- new CarbonTableIdentifier(dbName, tableName, ""),
+ absoluteTableIdentifier,
LockUsage.TABLE_STATUS_LOCK
)
// Delete marked loads
val isUpdationRequired =
DeleteLoadFolders.deleteLoadFoldersFromFileSystem(
- dbName,
- tableName,
- storePath,
+ absoluteTableIdentifier,
isForceDeletion,
details
)
@@ -212,8 +205,10 @@ object DataManagementFunc {
val latestStatus = CarbonLoaderUtil
.updateLoadMetadataFromOldToNew(details, latestMetadata)
- CarbonLoaderUtil.writeLoadMetadata(storePath, dbName, tableName, latestStatus)
+ CarbonLoaderUtil.writeLoadMetadata(absoluteTableIdentifier, latestStatus)
} else {
+ val dbName = absoluteTableIdentifier.getCarbonTableIdentifier.getDatabaseName
+ val tableName = absoluteTableIdentifier.getCarbonTableIdentifier.getTableName
val errorMsg = "Clean files request is failed for " +
s"$dbName.$tableName" +
". Not able to acquire the table status lock due to other operation " +
http://git-wip-us.apache.org/repos/asf/carbondata/blob/1155d4d8/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/PartitionDropper.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/PartitionDropper.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/PartitionDropper.scala
index 0a41f44..a82ea00 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/PartitionDropper.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/PartitionDropper.scala
@@ -42,8 +42,7 @@ object PartitionDropper {
val carbonTable = dropPartitionCallableModel.carbonTable
val dbName = carbonTable.getDatabaseName
val tableName = carbonTable.getFactTableName
- val identifier = carbonTable.getAbsoluteTableIdentifier
- val carbonTableIdentifier = identifier.getCarbonTableIdentifier
+ val absoluteTableIdentifier = carbonTable.getAbsoluteTableIdentifier
val partitionInfo = carbonTable.getPartitionInfo(tableName)
val partitioner = PartitionFactory.getPartitioner(partitionInfo)
@@ -68,7 +67,7 @@ object PartitionDropper {
for (i <- 0 until bucketNumber) {
val bucketId = i
val rdd = new CarbonScanPartitionRDD(alterPartitionModel,
- carbonTableIdentifier,
+ absoluteTableIdentifier,
Seq(partitionId, targetPartitionId),
bucketId
).partitionBy(partitioner).map(_._2)
@@ -77,7 +76,7 @@ object PartitionDropper {
new AlterPartitionResultImpl(),
Seq(partitionId),
bucketId,
- identifier,
+ absoluteTableIdentifier,
rdd).collect()
if (dropStatus.length == 0) {
@@ -95,7 +94,7 @@ object PartitionDropper {
if (finalDropStatus) {
try {
- PartitionUtils.deleteOriginalCarbonFile(alterPartitionModel, identifier,
+ PartitionUtils.deleteOriginalCarbonFile(alterPartitionModel, absoluteTableIdentifier,
Seq(partitionId, targetPartitionId).toList, dbName,
tableName, partitionInfo)
} catch {
@@ -111,7 +110,7 @@ object PartitionDropper {
case e: Exception => sys.error(s"Exception in dropping partition action: ${ e.getMessage }")
}
} else {
- PartitionUtils.deleteOriginalCarbonFile(alterPartitionModel, identifier,
+ PartitionUtils.deleteOriginalCarbonFile(alterPartitionModel, absoluteTableIdentifier,
Seq(partitionId).toList, dbName, tableName, partitionInfo)
logger.audit(s"Drop Partition request completed for table " +
s"${ dbName }.${ tableName }")
http://git-wip-us.apache.org/repos/asf/carbondata/blob/1155d4d8/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/PartitionSplitter.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/PartitionSplitter.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/PartitionSplitter.scala
index fca7542..db664b3 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/PartitionSplitter.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/PartitionSplitter.scala
@@ -39,8 +39,7 @@ object PartitionSplitter {
val partitionId = splitPartitionCallableModel.partitionId
val carbonLoadModel = splitPartitionCallableModel.carbonLoadModel
val carbonTable = carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable
- val identifier = carbonTable.getAbsoluteTableIdentifier
- val carbonTableIdentifier = identifier.getCarbonTableIdentifier
+ val absoluteTableIdentifier = carbonTable.getAbsoluteTableIdentifier
val tableName = carbonTable.getFactTableName
val databaseName = carbonTable.getDatabaseName
val bucketInfo = carbonTable.getBucketingInfo(tableName)
@@ -56,7 +55,7 @@ object PartitionSplitter {
val bucketId = i
val rdd = new CarbonScanPartitionRDD(
alterPartitionModel,
- carbonTableIdentifier,
+ absoluteTableIdentifier,
Seq(partitionId),
bucketId
).partitionBy(partitioner).map(_._2)
@@ -65,7 +64,7 @@ object PartitionSplitter {
new AlterPartitionResultImpl(),
Seq(partitionId),
bucketId,
- identifier,
+ absoluteTableIdentifier,
rdd).collect()
if (splitStatus.length == 0) {
@@ -83,8 +82,8 @@ object PartitionSplitter {
if (finalSplitStatus) {
try {
PartitionUtils.
- deleteOriginalCarbonFile(alterPartitionModel, identifier, Seq(partitionId).toList
- , databaseName, tableName, partitionInfo)
+ deleteOriginalCarbonFile(alterPartitionModel, absoluteTableIdentifier,
+ Seq(partitionId).toList, databaseName, tableName, partitionInfo)
} catch {
case e: IOException => sys.error(s"Exception while delete original carbon files " +
e.getMessage)
http://git-wip-us.apache.org/repos/asf/carbondata/blob/1155d4d8/integration/spark-common/src/main/scala/org/apache/carbondata/spark/tasks/DictionaryWriterTask.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/tasks/DictionaryWriterTask.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/tasks/DictionaryWriterTask.scala
index a9ac9f1..e304d84 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/tasks/DictionaryWriterTask.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/tasks/DictionaryWriterTask.scala
@@ -22,7 +22,7 @@ import scala.collection.mutable
import org.apache.carbondata.core.cache.dictionary.{Dictionary, DictionaryColumnUniqueIdentifier}
import org.apache.carbondata.core.constants.CarbonCommonConstants
-import org.apache.carbondata.core.metadata.{CarbonTableIdentifier, ColumnIdentifier}
+import org.apache.carbondata.core.metadata.{AbsoluteTableIdentifier, CarbonTableIdentifier, ColumnIdentifier}
import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema
import org.apache.carbondata.core.service.CarbonCommonFactory
import org.apache.carbondata.core.util.DataTypeUtil
@@ -32,18 +32,14 @@ import org.apache.carbondata.core.writer.CarbonDictionaryWriter
*
* @param valuesBuffer
* @param dictionary
- * @param carbonTableIdentifier
* @param dictionaryColumnUniqueIdentifier
- * @param carbonStoreLocation
* @param columnSchema
* @param isDictionaryFileExist
* @param writer
*/
class DictionaryWriterTask(valuesBuffer: mutable.HashSet[String],
dictionary: Dictionary,
- carbonTableIdentifier: CarbonTableIdentifier,
dictionaryColumnUniqueIdentifier: DictionaryColumnUniqueIdentifier,
- carbonStoreLocation: String,
columnSchema: ColumnSchema,
isDictionaryFileExist: Boolean,
var writer: CarbonDictionaryWriter = null) {
@@ -57,10 +53,7 @@ class DictionaryWriterTask(valuesBuffer: mutable.HashSet[String],
val values = valuesBuffer.toArray
java.util.Arrays.sort(values, Ordering[String])
val dictService = CarbonCommonFactory.getDictionaryService
- writer = dictService.getDictionaryWriter(
- carbonTableIdentifier,
- dictionaryColumnUniqueIdentifier,
- carbonStoreLocation)
+ writer = dictService.getDictionaryWriter(dictionaryColumnUniqueIdentifier)
val distinctValues: java.util.List[String] = new java.util.ArrayList()
try {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/1155d4d8/integration/spark-common/src/main/scala/org/apache/carbondata/spark/tasks/SortIndexWriterTask.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/tasks/SortIndexWriterTask.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/tasks/SortIndexWriterTask.scala
index 27f9418..f212120 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/tasks/SortIndexWriterTask.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/tasks/SortIndexWriterTask.scala
@@ -25,19 +25,15 @@ import org.apache.carbondata.core.writer.sortindex.{CarbonDictionarySortIndexWri
/**
* This task writes sort index file
*
- * @param carbonTableIdentifier
* @param dictionaryColumnUniqueIdentifier
* @param dataType
- * @param carbonStoreLocation
* @param dictionary
* @param distinctValues
* @param carbonDictionarySortIndexWriter
*/
class SortIndexWriterTask(
- carbonTableIdentifier: CarbonTableIdentifier,
dictionaryColumnUniqueIdentifier: DictionaryColumnUniqueIdentifier,
dataType: DataType,
- carbonStoreLocation: String,
dictionary: Dictionary,
distinctValues: java.util.List[String],
var carbonDictionarySortIndexWriter: CarbonDictionarySortIndexWriter = null) {
@@ -51,8 +47,7 @@ class SortIndexWriterTask(
dataType)
carbonDictionarySortIndexWriter =
dictService
- .getDictionarySortIndexWriter(carbonTableIdentifier, dictionaryColumnUniqueIdentifier,
- carbonStoreLocation)
+ .getDictionarySortIndexWriter(dictionaryColumnUniqueIdentifier)
carbonDictionarySortIndexWriter.writeSortIndex(dictionarySortInfo.getSortIndex)
carbonDictionarySortIndexWriter
.writeInvertedSortIndex(dictionarySortInfo.getSortIndexInverted)
http://git-wip-us.apache.org/repos/asf/carbondata/blob/1155d4d8/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/CommonUtil.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/CommonUtil.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/CommonUtil.scala
index b51d0f0..1b21e3d 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/CommonUtil.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/CommonUtil.scala
@@ -27,6 +27,7 @@ import scala.collection.mutable.Map
import org.apache.commons.lang3.StringUtils
import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat
import org.apache.spark.SparkContext
import org.apache.spark.sql.{Row, RowFactory, SQLContext}
@@ -40,7 +41,7 @@ import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.datastore.filesystem.CarbonFile
import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.memory.{UnsafeMemoryManager, UnsafeSortMemoryManager}
-import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier
+import org.apache.carbondata.core.metadata.{AbsoluteTableIdentifier, CarbonTableIdentifier}
import org.apache.carbondata.core.metadata.datatype.{DataType, DataTypes}
import org.apache.carbondata.core.metadata.schema.PartitionInfo
import org.apache.carbondata.core.metadata.schema.partition.PartitionType
@@ -54,6 +55,7 @@ import org.apache.carbondata.core.util.path.CarbonStorePath
import org.apache.carbondata.processing.loading.csvinput.CSVInputFormat
import org.apache.carbondata.processing.loading.exception.CarbonDataLoadingException
import org.apache.carbondata.processing.loading.model.CarbonLoadModel
+import org.apache.carbondata.processing.merger.TableMeta
import org.apache.carbondata.processing.util.{CarbonDataProcessorUtil, CarbonLoaderUtil}
import org.apache.carbondata.spark.exception.MalformedCarbonCommandException
import org.apache.carbondata.spark.rdd.CarbonMergeFilesRDD
@@ -848,14 +850,32 @@ object CommonUtil {
*/
def mergeIndexFiles(sparkContext: SparkContext,
segmentIds: Seq[String],
- storePath: String,
+ tablePath: String,
carbonTable: CarbonTable): Unit = {
if (CarbonProperties.getInstance().getProperty(
CarbonCommonConstants.CARBON_MERGE_INDEX_IN_SEGMENT,
CarbonCommonConstants.CARBON_MERGE_INDEX_IN_SEGMENT_DEFAULT).toBoolean) {
- new CarbonMergeFilesRDD(sparkContext, AbsoluteTableIdentifier.from(storePath,
+ new CarbonMergeFilesRDD(sparkContext, AbsoluteTableIdentifier.from(tablePath,
carbonTable.getDatabaseName, carbonTable.getFactTableName).getTablePath,
segmentIds).collect()
}
}
+
+ /**
+ * can be removed with the spark 1.6 removal
+ * @param tableMeta
+ * @return
+ */
+ @deprecated
+ def getTablePath(tableMeta: TableMeta): String = {
+ if (tableMeta.tablePath == null) {
+ tableMeta.storePath + CarbonCommonConstants.FILE_SEPARATOR +
+ tableMeta.carbonTableIdentifier.getDatabaseName +
+ CarbonCommonConstants.FILE_SEPARATOR + tableMeta.carbonTableIdentifier.getTableName
+ }
+ else {
+ tableMeta.tablePath
+ }
+ }
+
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/1155d4d8/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/DataLoadingUtil.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/DataLoadingUtil.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/DataLoadingUtil.scala
index 5a24d7d..35e1e78 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/DataLoadingUtil.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/DataLoadingUtil.scala
@@ -182,7 +182,7 @@ object DataLoadingUtil {
carbonLoadModel: CarbonLoadModel): Unit = {
carbonLoadModel.setTableName(table.getFactTableName)
carbonLoadModel.setDatabaseName(table.getDatabaseName)
- carbonLoadModel.setStorePath(table.getStorePath)
+ carbonLoadModel.setTablePath(table.getTablePath)
carbonLoadModel.setTableName(table.getFactTableName)
val dataLoadSchema = new CarbonDataLoadSchema(table)
// Need to fill dimension relation
http://git-wip-us.apache.org/repos/asf/carbondata/blob/1155d4d8/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtil.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtil.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtil.scala
index 3bb8d94..975fc9b 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtil.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtil.scala
@@ -42,7 +42,7 @@ import org.apache.carbondata.core.cache.dictionary.{Dictionary, DictionaryColumn
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.locks.{CarbonLockFactory, LockUsage}
-import org.apache.carbondata.core.metadata.{CarbonTableIdentifier, ColumnIdentifier}
+import org.apache.carbondata.core.metadata.{AbsoluteTableIdentifier, CarbonTableIdentifier, ColumnIdentifier}
import org.apache.carbondata.core.metadata.datatype.{DataType, DataTypes}
import org.apache.carbondata.core.metadata.encoder.Encoding
import org.apache.carbondata.core.metadata.schema.table.column.{CarbonDimension, ColumnSchema}
@@ -174,12 +174,9 @@ object GlobalDictionaryUtil {
model.table,
model.columnIdentifier(columnIndex),
model.columnIdentifier(columnIndex).getDataType,
- CarbonStorePath.getCarbonTablePath(model.hdfsLocation, model.table))
- val writer: CarbonDictionaryWriter = dictService.getDictionaryWriter(
- model.table,
- dictionaryColumnUniqueIdentifier,
- model.hdfsLocation
- )
+ CarbonStorePath.getCarbonTablePath(model.table))
+ val writer: CarbonDictionaryWriter = dictService
+ .getDictionaryWriter(dictionaryColumnUniqueIdentifier)
try {
while (iter.hasNext) {
writer.write(iter.next)
@@ -196,8 +193,7 @@ object GlobalDictionaryUtil {
val dictMap = new HashMap[String, Dictionary]
model.primDimensions.zipWithIndex.filter(f => model.dictFileExists(f._2)).foreach { m =>
val dict = CarbonLoaderUtil.getDictionary(model.table,
- m._1.getColumnIdentifier, model.hdfsLocation,
- m._1.getDataType
+ m._1.getColumnIdentifier, m._1.getDataType
)
dictMap.put(m._1.getColumnId, dict)
}
@@ -218,12 +214,11 @@ object GlobalDictionaryUtil {
model.table,
model.columnIdentifier(i),
model.columnIdentifier(i).getDataType,
- CarbonStorePath.getCarbonTablePath(model.hdfsLocation, model.table))
+ CarbonStorePath.getCarbonTablePath(model.table))
val set = new HashSet[String]
if (model.dictFileExists(i)) {
- val reader: CarbonDictionaryReader = dictService.getDictionaryReader(model.table,
- dictionaryColumnUniqueIdentifier, model.hdfsLocation
- )
+ val reader: CarbonDictionaryReader = dictService.getDictionaryReader(
+ dictionaryColumnUniqueIdentifier)
val values = reader.read
if (values != null) {
for (j <- 0 until values.size) {
@@ -330,7 +325,8 @@ object GlobalDictionaryUtil {
if (null == carbonLoadModel.getLoadMetadataDetails) {
CommonUtil.readLoadMetadataDetails(carbonLoadModel)
}
- DictionaryLoadModel(table,
+ val absoluteTableIdentifier = new AbsoluteTableIdentifier(hdfsLocation, table)
+ DictionaryLoadModel(absoluteTableIdentifier,
dimensions,
hdfsLocation,
dictfolderPath,
@@ -391,7 +387,7 @@ object GlobalDictionaryUtil {
model: DictionaryLoadModel,
status: Array[(Int, SegmentStatus)]) = {
var result = false
- val tableName = model.table.getTableName
+ val tableName = model.table.getCarbonTableIdentifier.getTableName
status.foreach { x =>
val columnName = model.primDimensions(x._1).getColName
if (SegmentStatus.LOAD_FAILURE == x._2) {
@@ -819,21 +815,19 @@ object GlobalDictionaryUtil {
*
* @param carbonTablePath
* @param columnSchema
- * @param tableIdentifier
- * @param storePath
+ * @param absoluteTableIdentifier
* @param defaultValue
*/
def loadDefaultDictionaryValueForNewColumn(carbonTablePath: CarbonTablePath,
columnSchema: ColumnSchema,
- tableIdentifier: CarbonTableIdentifier,
- storePath: String,
+ absoluteTableIdentifier: AbsoluteTableIdentifier,
defaultValue: String): Unit = {
var carbonDictionarySortIndexWriter: CarbonDictionarySortIndexWriter = null
var dictionary: Dictionary = null
val dictLock = CarbonLockFactory
- .getCarbonLockObj(carbonTablePath.getRelativeDictionaryDirectory,
+ .getCarbonLockObj(absoluteTableIdentifier,
columnSchema.getColumnUniqueId + LockUsage.LOCK)
var isDictionaryLocked = false
try {
@@ -852,7 +846,7 @@ object GlobalDictionaryUtil {
columnSchema.getDataType)
val dictionaryColumnUniqueIdentifier: DictionaryColumnUniqueIdentifier = new
DictionaryColumnUniqueIdentifier(
- tableIdentifier,
+ absoluteTableIdentifier,
columnIdentifier,
columnIdentifier.getDataType,
carbonTablePath)
@@ -863,9 +857,7 @@ object GlobalDictionaryUtil {
}
val dictWriteTask = new DictionaryWriterTask(valuesBuffer,
dictionary,
- tableIdentifier,
dictionaryColumnUniqueIdentifier,
- storePath,
columnSchema,
false
)
@@ -875,10 +867,9 @@ object GlobalDictionaryUtil {
}")
if (distinctValues.size() > 0) {
- val sortIndexWriteTask = new SortIndexWriterTask(tableIdentifier,
+ val sortIndexWriteTask = new SortIndexWriterTask(
dictionaryColumnUniqueIdentifier,
columnSchema.getDataType,
- storePath,
dictionary,
distinctValues)
sortIndexWriteTask.execute()
http://git-wip-us.apache.org/repos/asf/carbondata/blob/1155d4d8/integration/spark-common/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchemaCommon.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchemaCommon.scala b/integration/spark-common/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchemaCommon.scala
index a2f3364..756de6b 100644
--- a/integration/spark-common/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchemaCommon.scala
+++ b/integration/spark-common/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchemaCommon.scala
@@ -103,8 +103,7 @@ case class DataLoadTableFileMapping(table: String, loadPath: String)
case class ExecutionErrors(var failureCauses: FailureCauses, var errorMsg: String )
-case class CarbonMergerMapping(storeLocation: String,
- hdfsStoreLocation: String,
+case class CarbonMergerMapping(hdfsStoreLocation: String,
metadataFilePath: String,
var mergedLoadName: String,
databaseName: String,
@@ -136,7 +135,6 @@ case class CompactionModel(compactionSize: Long,
isDDLTrigger: Boolean)
case class CompactionCallableModel(carbonLoadModel: CarbonLoadModel,
- storeLocation: String,
carbonTable: CarbonTable,
loadsToMerge: util.List[LoadMetadataDetails],
sqlContext: SQLContext,
@@ -203,7 +201,7 @@ class AlterTableColumnSchemaGenerator(
tableInfo: TableInfo,
carbonTablePath: CarbonTablePath,
tableIdentifier: CarbonTableIdentifier,
- storePath: String, sc: SparkContext) {
+ sc: SparkContext) {
val LOGGER = LogServiceFactory.getLogService(this.getClass.getName)
http://git-wip-us.apache.org/repos/asf/carbondata/blob/1155d4d8/integration/spark-common/src/main/scala/org/apache/spark/sql/test/TestQueryExecutor.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/spark/sql/test/TestQueryExecutor.scala b/integration/spark-common/src/main/scala/org/apache/spark/sql/test/TestQueryExecutor.scala
index d2b565f..4a91b47 100644
--- a/integration/spark-common/src/main/scala/org/apache/spark/sql/test/TestQueryExecutor.scala
+++ b/integration/spark-common/src/main/scala/org/apache/spark/sql/test/TestQueryExecutor.scala
@@ -51,6 +51,7 @@ object TestQueryExecutor {
LOGGER.info(s"project path: $projectPath")
val integrationPath = s"$projectPath/integration"
val metastoredb = s"$integrationPath/spark-common/target"
+ val location = s"$integrationPath/spark-common/target/dbpath"
val masterUrl = {
val property = System.getProperty("spark.master.url")
if (property == null) {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/1155d4d8/integration/spark-common/src/main/scala/org/apache/spark/sql/test/util/QueryTest.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/spark/sql/test/util/QueryTest.scala b/integration/spark-common/src/main/scala/org/apache/spark/sql/test/util/QueryTest.scala
index 0a4231c..c01cfef 100644
--- a/integration/spark-common/src/main/scala/org/apache/spark/sql/test/util/QueryTest.scala
+++ b/integration/spark-common/src/main/scala/org/apache/spark/sql/test/util/QueryTest.scala
@@ -105,6 +105,7 @@ class QueryTest extends PlanTest {
getProperty(CarbonCommonConstants.STORE_LOCATION)
val resourcesPath = TestQueryExecutor.resourcesPath
val integrationPath = TestQueryExecutor.integrationPath
+ val dblocation = TestQueryExecutor.location
}
object QueryTest {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/1155d4d8/integration/spark-common/src/main/scala/org/apache/spark/util/PartitionUtils.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/spark/util/PartitionUtils.scala b/integration/spark-common/src/main/scala/org/apache/spark/util/PartitionUtils.scala
index dcda72b..82052aa 100644
--- a/integration/spark-common/src/main/scala/org/apache/spark/util/PartitionUtils.scala
+++ b/integration/spark-common/src/main/scala/org/apache/spark/util/PartitionUtils.scala
@@ -29,7 +29,7 @@ import org.apache.hadoop.mapreduce.Job
import org.apache.spark.sql.execution.command.AlterPartitionModel
import org.apache.carbondata.core.datastore.block.{SegmentProperties, TableBlockInfo}
-import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier
+import org.apache.carbondata.core.metadata.{AbsoluteTableIdentifier, CarbonTableIdentifier}
import org.apache.carbondata.core.metadata.schema.PartitionInfo
import org.apache.carbondata.core.metadata.schema.partition.PartitionType
import org.apache.carbondata.core.util.CarbonUtil
@@ -160,12 +160,13 @@ object PartitionUtils {
val segmentId = alterPartitionModel.segmentId
val oldPartitionIds = alterPartitionModel.oldPartitionIds
val newTime = carbonLoadModel.getFactTimeStamp
- val storePath = carbonLoadModel.getStorePath
+ val tablePath = carbonLoadModel.getTablePath
val tableBlockInfoList =
getPartitionBlockList(identifier, segmentId, partitionIds, oldPartitionIds,
partitionInfo).asScala
val pathList: util.List[String] = new util.ArrayList[String]()
- val carbonTablePath = new CarbonTablePath(storePath, dbName, tableName)
+ val carbonTableIdentifier = new CarbonTableIdentifier(dbName, tableName, "")
+ val carbonTablePath = new CarbonTablePath(carbonTableIdentifier, tablePath)
tableBlockInfoList.foreach{ tableBlockInfo =>
val path = tableBlockInfo.getFilePath
val timestamp = CarbonTablePath.DataFileUtil.getTimeStampFromFileName(path)
http://git-wip-us.apache.org/repos/asf/carbondata/blob/1155d4d8/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 d7986dc..7c5599b 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
@@ -50,7 +50,7 @@ import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.mutate.CarbonUpdateUtil
import org.apache.carbondata.core.scan.partition.PartitionUtil
import org.apache.carbondata.core.statusmanager.{LoadMetadataDetails, SegmentStatus, SegmentStatusManager}
-import org.apache.carbondata.core.util.{ByteUtil, CarbonProperties}
+import org.apache.carbondata.core.util.{ByteUtil, CarbonProperties, CarbonUtil}
import org.apache.carbondata.core.util.path.CarbonStorePath
import org.apache.carbondata.events.{LoadTablePostExecutionEvent, OperationContext, OperationListenerBus}
import org.apache.carbondata.processing.exception.DataLoadingException
@@ -82,10 +82,15 @@ object CarbonDataRDDFactory {
compactionType: CompactionType,
carbonTable: CarbonTable,
compactionModel: CompactionModel): Unit = {
+ // taking system level lock at the mdt file location
+ var configuredMdtPath = CarbonProperties.getInstance()
+ .getProperty(CarbonCommonConstants.CARBON_UPDATE_SYNC_FOLDER,
+ CarbonCommonConstants.CARBON_UPDATE_SYNC_FOLDER_DEFAULT).trim
+ configuredMdtPath = CarbonUtil.checkAndAppendFileSystemURIScheme(configuredMdtPath)
val lock = CarbonLockFactory
- .getCarbonLockObj(CarbonCommonConstants.SYSTEM_LEVEL_COMPACTION_LOCK_FOLDER,
- LockUsage.SYSTEMLEVEL_COMPACTION_LOCK
- )
+ .getCarbonLockObj(configuredMdtPath + CarbonCommonConstants.FILE_SEPARATOR +
+ CarbonCommonConstants.SYSTEM_LEVEL_COMPACTION_LOCK_FOLDER,
+ LockUsage.SYSTEMLEVEL_COMPACTION_LOCK)
if (lock.lockWithRetries()) {
LOGGER.info(s"Acquired the compaction lock for table ${ carbonLoadModel.getDatabaseName }" +
s".${ carbonLoadModel.getTableName }")
@@ -249,7 +254,7 @@ object CarbonDataRDDFactory {
loadModel.setCarbonDataLoadSchema(dataLoadSchema)
loadModel.setTableName(table.getCarbonTableIdentifier.getTableName)
loadModel.setDatabaseName(table.getCarbonTableIdentifier.getDatabaseName)
- loadModel.setStorePath(table.getStorePath)
+ loadModel.setTablePath(table.getTablePath)
CommonUtil.readLoadMetadataDetails(loadModel)
val loadStartTime = CarbonUpdateUtil.readCurrentTime()
loadModel.setFactTimeStamp(loadStartTime)
@@ -273,8 +278,7 @@ object CarbonDataRDDFactory {
LOGGER.audit(s"Data load request has been received for table" +
s" ${ carbonLoadModel.getDatabaseName }.${ carbonLoadModel.getTableName }")
// Check if any load need to be deleted before loading new data
- DataManagementFunc.deleteLoadsAndUpdateMetadata(carbonLoadModel.getDatabaseName,
- carbonLoadModel.getTableName, storePath, isForceDeletion = false, carbonTable)
+ DataManagementFunc.deleteLoadsAndUpdateMetadata(isForceDeletion = false, carbonTable)
var status: Array[(String, (LoadMetadataDetails, ExecutionErrors))] = null
var res: Array[List[(String, (LoadMetadataDetails, ExecutionErrors))]] = null
@@ -531,7 +535,7 @@ object CarbonDataRDDFactory {
carbonTable.getMetaDataFilepath)
val segmentIds = loadMetadataDetails.map(_.getLoadName)
val segmentIdIndex = segmentIds.zipWithIndex.toMap
- val carbonTablePath = CarbonStorePath.getCarbonTablePath(carbonLoadModel.getStorePath,
+ val carbonTablePath = CarbonStorePath.getCarbonTablePath(carbonLoadModel.getTablePath,
carbonTable.getCarbonTableIdentifier)
val segmentId2maxTaskNo = segmentIds.map { segId =>
(segId, CarbonUpdateUtil.getLatestTaskIdForSegment(segId, carbonTablePath))
@@ -698,7 +702,7 @@ object CarbonDataRDDFactory {
)
} else {
val lock = CarbonLockFactory.getCarbonLockObj(
- carbonTable.getAbsoluteTableIdentifier.getCarbonTableIdentifier,
+ carbonTable.getAbsoluteTableIdentifier,
LockUsage.COMPACTION_LOCK)
if (lock.lockWithRetries()) {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/1155d4d8/integration/spark2/src/main/scala/org/apache/carbondata/spark/util/CarbonSparkUtil.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/carbondata/spark/util/CarbonSparkUtil.scala b/integration/spark2/src/main/scala/org/apache/carbondata/spark/util/CarbonSparkUtil.scala
index 5dd5983..1e6a36e 100644
--- a/integration/spark2/src/main/scala/org/apache/carbondata/spark/util/CarbonSparkUtil.scala
+++ b/integration/spark2/src/main/scala/org/apache/carbondata/spark/util/CarbonSparkUtil.scala
@@ -46,10 +46,9 @@ object CarbonSparkUtil {
}
def createCarbonRelation(tableInfo: TableInfo, tablePath: String): CarbonRelation = {
- val identifier = AbsoluteTableIdentifier.fromTablePath(tablePath)
val table = CarbonTable.buildFromTableInfo(tableInfo)
- val meta = new TableMeta(identifier.getCarbonTableIdentifier,
- identifier.getStorePath, tablePath, table)
+ val meta = new TableMeta(table.getCarbonTableIdentifier,
+ table.getTablePath, tablePath, table)
CarbonRelation(tableInfo.getDatabaseName, tableInfo.getFactTable.getTableName,
CarbonSparkUtil.createSparkMeta(table), meta)
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/1155d4d8/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonCatalystOperators.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonCatalystOperators.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonCatalystOperators.scala
index 4acb82c..48f1a09 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonCatalystOperators.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonCatalystOperators.scala
@@ -25,6 +25,9 @@ import org.apache.spark.sql.hive.HiveSessionCatalog
import org.apache.spark.sql.optimizer.CarbonDecoderRelation
import org.apache.spark.sql.types.{StringType, TimestampType}
+import org.apache.carbondata.core.constants.CarbonCommonConstants
+import org.apache.carbondata.core.datastore.impl.FileFactory
+import org.apache.carbondata.core.util.CarbonProperties
import org.apache.carbondata.spark.CarbonAliasDecoderRelation
case class CarbonDictionaryCatalystDecoder(
@@ -70,6 +73,40 @@ object GetDB {
dbName.getOrElse(
sparkSession.sessionState.catalog.asInstanceOf[HiveSessionCatalog].getCurrentDatabase)
}
+
+ /**
+ * The method returns the database location
+ * if carbon.storeLocation does point to spark.sql.warehouse.dir then returns
+ * the database locationUri as database location else follows the old behaviour
+ * making database location from carbon fixed store and database name.
+ *
+ * @param dbName
+ * @param sparkSession
+ * @param fixedStorePath
+ * @return
+ */
+ def getDatabaseLocation(dbName: String, sparkSession: SparkSession,
+ fixedStorePath: String): String = {
+ var databaseLocation =
+ sparkSession.sessionState.catalog.asInstanceOf[HiveSessionCatalog].getDatabaseMetadata(dbName)
+ .locationUri
+ // for default database and db ends with .db
+ // check whether the carbon store and hive store is same or different.
+ if (dbName.equals("default") || databaseLocation.endsWith(".db")) {
+ val properties = CarbonProperties.getInstance()
+ val carbonStorePath = FileFactory
+ .getUpdatedFilePath(properties.getProperty(CarbonCommonConstants.STORE_LOCATION))
+ val hiveStorePath = FileFactory
+ .getUpdatedFilePath(sparkSession.conf.get("spark.sql.warehouse.dir"))
+ // if carbon.store does not point to spark.sql.warehouse.dir then follow the old table path
+ // format
+ if (!hiveStorePath.equals(carbonStorePath)) {
+ databaseLocation = fixedStorePath + CarbonCommonConstants.FILE_SEPARATOR + dbName
+ }
+ }
+
+ return FileFactory.getUpdatedFilePath(databaseLocation)
+ }
}
/**
http://git-wip-us.apache.org/repos/asf/carbondata/blob/1155d4d8/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDatasourceHadoopRelation.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDatasourceHadoopRelation.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDatasourceHadoopRelation.scala
index 3fb65be..22933f2 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDatasourceHadoopRelation.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDatasourceHadoopRelation.scala
@@ -47,7 +47,8 @@ case class CarbonDatasourceHadoopRelation(
isSubquery: ArrayBuffer[Boolean] = new ArrayBuffer[Boolean]())
extends BaseRelation with InsertableRelation {
- lazy val identifier: AbsoluteTableIdentifier = AbsoluteTableIdentifier.fromTablePath(paths.head)
+ lazy val identifier: AbsoluteTableIdentifier = AbsoluteTableIdentifier.from(paths.head,
+ parameters("dbname"), parameters("tablename"))
lazy val databaseName: String = carbonTable.getDatabaseName
lazy val tableName: String = carbonTable.getFactTableName
CarbonSession.updateSessionInfoToCurrentThread(sparkSession)