You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ja...@apache.org on 2018/01/08 14:32:33 UTC
[1/7] carbondata git commit: [CARBONDATA-1976][PARTITION] Support
combination of dynamic and static partitions. And fix concurrent partition
load issue.
Repository: carbondata
Updated Branches:
refs/heads/carbonstore 829e7aa4e -> 90921eb7b
[CARBONDATA-1976][PARTITION] Support combination of dynamic and static partitions. And fix concurrent partition load issue.
Support combination of dynamic and static partitions.
This closes #1755
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/af4277e9
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/af4277e9
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/af4277e9
Branch: refs/heads/carbonstore
Commit: af4277e9495faac3082e8c179cc049bcbeb699e2
Parents: 829e7aa
Author: ravipesala <ra...@gmail.com>
Authored: Wed Jan 3 10:48:09 2018 +0530
Committer: Jacky Li <ja...@qq.com>
Committed: Sun Jan 7 00:01:24 2018 +0800
----------------------------------------------------------------------
.../core/metadata/PartitionMapFileStore.java | 40 +++++-
.../hadoop/api/CarbonOutputCommitter.java | 8 +-
.../StandardPartitionTableDropTestCase.scala | 28 +++++
.../StandardPartitionTableLoadingTestCase.scala | 49 +++++++-
...tandardPartitionTableOverwriteTestCase.scala | 74 +++++++++++
.../spark/rdd/CarbonDropPartitionRDD.scala | 8 +-
.../spark/sql/catalyst/CarbonDDLSqlParser.scala | 6 +-
.../management/CarbonLoadDataCommand.scala | 122 +++++++++++--------
...rbonAlterTableDropHivePartitionCommand.scala | 19 ++-
.../datasources/CarbonFileFormat.scala | 48 ++++++--
.../sql/parser/CarbonSpark2SqlParser.scala | 4 +-
11 files changed, 327 insertions(+), 79 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/af4277e9/core/src/main/java/org/apache/carbondata/core/metadata/PartitionMapFileStore.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/metadata/PartitionMapFileStore.java b/core/src/main/java/org/apache/carbondata/core/metadata/PartitionMapFileStore.java
index d29dfbb..f7074c4 100644
--- a/core/src/main/java/org/apache/carbondata/core/metadata/PartitionMapFileStore.java
+++ b/core/src/main/java/org/apache/carbondata/core/metadata/PartitionMapFileStore.java
@@ -237,15 +237,21 @@ public class PartitionMapFileStore {
* @param segmentPath
* @param partitionsToDrop
* @param uniqueId
+ * @param partialMatch If it is true then even the partial partition spec matches also can be
+ * dropped
* @throws IOException
*/
- public void dropPartitions(String segmentPath, List<String> partitionsToDrop, String uniqueId)
- throws IOException {
+ public void dropPartitions(String segmentPath, List<String> partitionsToDrop, String uniqueId,
+ boolean partialMatch) throws IOException {
readAllPartitionsOfSegment(segmentPath);
List<String> indexesToDrop = new ArrayList<>();
for (Map.Entry<String, List<String>> entry: partitionMap.entrySet()) {
- for (String partition: partitionsToDrop) {
- if (entry.getValue().contains(partition)) {
+ if (partialMatch) {
+ if (entry.getValue().containsAll(partitionsToDrop)) {
+ indexesToDrop.add(entry.getKey());
+ }
+ } else {
+ if (partitionsToDrop.containsAll(entry.getValue())) {
indexesToDrop.add(entry.getKey());
}
}
@@ -302,7 +308,7 @@ public class PartitionMapFileStore {
LoadMetadataDetails[] details = ssm.readLoadMetadata(table.getMetaDataFilepath());
// scan through each segment.
-
+ List<String> segmentsNeedToBeDeleted = new ArrayList<>();
for (LoadMetadataDetails segment : details) {
// if this segment is valid then only we will go for deletion of related
@@ -318,6 +324,12 @@ public class PartitionMapFileStore {
String partitionFilePath = getPartitionFilePath(segmentPath);
if (partitionFilePath != null) {
PartitionMapper partitionMapper = readPartitionMap(partitionFilePath);
+ if (partitionMapper.partitionMap.size() == 0) {
+ // There is no partition information, it means all partitions are dropped.
+ // So segment need to be marked as delete.
+ segmentsNeedToBeDeleted.add(segment.getLoadName());
+ continue;
+ }
DataFileFooterConverter fileFooterConverter = new DataFileFooterConverter();
SegmentIndexFileStore indexFileStore = new SegmentIndexFileStore();
indexFileStore.readAllIIndexOfSegment(segmentPath);
@@ -356,7 +368,11 @@ public class PartitionMapFileStore {
// Delete all old partition files
for (CarbonFile partitionFile : partitionFiles) {
if (!currentPartitionFile.getName().equalsIgnoreCase(partitionFile.getName())) {
- partitionFile.delete();
+ long fileTimeStamp = Long.parseLong(partitionFile.getName().substring(0,
+ partitionFile.getName().length() - CarbonTablePath.PARTITION_MAP_EXT.length()));
+ if (CarbonUpdateUtil.isMaxQueryTimeoutExceeded(fileTimeStamp) || forceDelete) {
+ partitionFile.delete();
+ }
}
}
}
@@ -370,6 +386,18 @@ public class PartitionMapFileStore {
}
}
}
+ // If any segments that are required to delete
+ if (segmentsNeedToBeDeleted.size() > 0) {
+ try {
+ // Mark the segments as delete.
+ SegmentStatusManager.updateDeletionStatus(
+ table.getAbsoluteTableIdentifier(),
+ segmentsNeedToBeDeleted,
+ table.getMetaDataFilepath());
+ } catch (Exception e) {
+ throw new IOException(e);
+ }
+ }
}
public List<String> getPartitions(String indexFileName) {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/af4277e9/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonOutputCommitter.java
----------------------------------------------------------------------
diff --git a/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonOutputCommitter.java b/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonOutputCommitter.java
index 525249a..bc7c56f 100644
--- a/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonOutputCommitter.java
+++ b/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonOutputCommitter.java
@@ -83,7 +83,13 @@ public class CarbonOutputCommitter extends FileOutputCommitter {
* @throws IOException
*/
@Override public void commitJob(JobContext context) throws IOException {
- super.commitJob(context);
+ try {
+ super.commitJob(context);
+ } catch (IOException e) {
+ // ignore, in case of concurrent load it try to remove temporary folders by other load may
+ // cause file not found exception. This will not impact carbon load,
+ LOGGER.warn(e.getMessage());
+ }
boolean overwriteSet = CarbonTableOutputFormat.isOverwriteSet(context.getConfiguration());
CarbonLoadModel loadModel = CarbonTableOutputFormat.getLoadModel(context.getConfiguration());
LoadMetadataDetails newMetaEntry = loadModel.getCurrentLoadMetadataDetail();
http://git-wip-us.apache.org/repos/asf/carbondata/blob/af4277e9/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/standardpartition/StandardPartitionTableDropTestCase.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/standardpartition/StandardPartitionTableDropTestCase.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/standardpartition/StandardPartitionTableDropTestCase.scala
index 9a9940b..2aa9145 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/standardpartition/StandardPartitionTableDropTestCase.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/standardpartition/StandardPartitionTableDropTestCase.scala
@@ -155,6 +155,33 @@ class StandardPartitionTableDropTestCase extends QueryTest with BeforeAndAfterAl
}
+
+ test("dropping all partition on table and do compaction") {
+ sql(
+ """
+ | CREATE TABLE partitionallcompaction (empno int, empname String, designation String,
+ | workgroupcategory int, workgroupcategoryname String, deptno int,
+ | projectjoindate Timestamp, projectenddate Date,attendance int,
+ | utilization int,salary int)
+ | PARTITIONED BY (deptname String,doj Timestamp,projectcode int)
+ | STORED BY 'org.apache.carbondata.format'
+ """.stripMargin)
+ sql(s"""LOAD DATA local inpath '$resourcesPath/data.csv' INTO TABLE partitionallcompaction OPTIONS('DELIMITER'= ',', 'QUOTECHAR'= '"')""")
+ sql(s"""LOAD DATA local inpath '$resourcesPath/data.csv' INTO TABLE partitionallcompaction OPTIONS('DELIMITER'= ',', 'QUOTECHAR'= '"')""")
+ sql(s"""LOAD DATA local inpath '$resourcesPath/data.csv' INTO TABLE partitionallcompaction OPTIONS('DELIMITER'= ',', 'QUOTECHAR'= '"')""")
+ sql(s"""LOAD DATA local inpath '$resourcesPath/data.csv' INTO TABLE partitionallcompaction OPTIONS('DELIMITER'= ',', 'QUOTECHAR'= '"')""")
+ sql(s"""ALTER TABLE partitionallcompaction DROP PARTITION(deptname='Learning')""")
+ sql(s"""ALTER TABLE partitionallcompaction DROP PARTITION(deptname='configManagement')""")
+ sql(s"""ALTER TABLE partitionallcompaction DROP PARTITION(deptname='network')""")
+ sql(s"""ALTER TABLE partitionallcompaction DROP PARTITION(deptname='protocol')""")
+ sql(s"""ALTER TABLE partitionallcompaction DROP PARTITION(deptname='security')""")
+ assert(sql(s"""SHOW PARTITIONS partitionallcompaction""").collect().length == 0)
+ sql("ALTER TABLE partitionallcompaction COMPACT 'MAJOR'").collect()
+ checkAnswer(
+ sql(s"""select count (*) from partitionallcompaction"""),
+ Seq(Row(0)))
+ }
+
override def afterAll = {
dropTable
}
@@ -167,6 +194,7 @@ class StandardPartitionTableDropTestCase extends QueryTest with BeforeAndAfterAl
sql("drop table if exists partitionmany")
sql("drop table if exists partitionshow")
sql("drop table if exists staticpartition")
+ sql("drop table if exists partitionallcompaction")
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/af4277e9/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/standardpartition/StandardPartitionTableLoadingTestCase.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/standardpartition/StandardPartitionTableLoadingTestCase.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/standardpartition/StandardPartitionTableLoadingTestCase.scala
index b399138..25e73c4 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/standardpartition/StandardPartitionTableLoadingTestCase.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/standardpartition/StandardPartitionTableLoadingTestCase.scala
@@ -17,6 +17,8 @@
package org.apache.carbondata.spark.testsuite.standardpartition
import java.io.{File, IOException}
+import java.util
+import java.util.concurrent.{Callable, ExecutorService, Executors}
import org.apache.commons.io.FileUtils
import org.apache.spark.sql.test.util.QueryTest
@@ -31,7 +33,7 @@ import org.apache.carbondata.core.util.CarbonProperties
import org.apache.carbondata.core.util.path.CarbonTablePath
class StandardPartitionTableLoadingTestCase extends QueryTest with BeforeAndAfterAll {
-
+ var executorService: ExecutorService = _
override def beforeAll {
dropTable
@@ -275,6 +277,47 @@ class StandardPartitionTableLoadingTestCase extends QueryTest with BeforeAndAfte
}
}
+ test("concurrent partition table load test") {
+ executorService = Executors.newCachedThreadPool()
+ sql(
+ """
+ | CREATE TABLE partitionmultiplethreeconcurrent (empno int, doj Timestamp,
+ | workgroupcategoryname String, deptno int, deptname String,
+ | projectcode int, projectjoindate Timestamp, projectenddate Timestamp,attendance int,
+ | utilization int,salary int)
+ | PARTITIONED BY (workgroupcategory int, empname String, designation String)
+ | STORED BY 'org.apache.carbondata.format'
+ | TBLPROPERTIES('DICTIONARY_INCLUDE'='empname,designation,deptname')
+ """.stripMargin)
+
+ val tasks = new util.ArrayList[Callable[String]]()
+ tasks.add(new QueryTask(s"""LOAD DATA local inpath '$resourcesPath/data.csv' INTO TABLE partitionmultiplethreeconcurrent OPTIONS('DELIMITER'= ',', 'QUOTECHAR'= '"')"""))
+ tasks.add(new QueryTask(s"""LOAD DATA local inpath '$resourcesPath/data.csv' INTO TABLE partitionmultiplethreeconcurrent OPTIONS('DELIMITER'= ',', 'QUOTECHAR'= '"')"""))
+ tasks.add(new QueryTask(s"""LOAD DATA local inpath '$resourcesPath/data.csv' INTO TABLE partitionmultiplethreeconcurrent OPTIONS('DELIMITER'= ',', 'QUOTECHAR'= '"')"""))
+ val results = executorService.invokeAll(tasks)
+ for (i <- 0 until tasks.size()) {
+ val res = results.get(i).get
+ assert("PASS".equals(res))
+ }
+ executorService.shutdown()
+ checkAnswer(sql("select count(*) from partitionmultiplethreeconcurrent"), Seq(Row(30)))
+ }
+
+ class QueryTask(query: String) extends Callable[String] {
+ override def call(): String = {
+ var result = "PASS"
+ try {
+ LOGGER.info("Executing :" + Thread.currentThread().getName)
+ sql(query)
+ } catch {
+ case ex: Exception =>
+ ex.printStackTrace()
+ result = "FAIL"
+ }
+ result
+ }
+ }
+
test("merge carbon index disable data loading for partition table for three partition column") {
CarbonProperties.getInstance.addProperty(
CarbonCommonConstants.CARBON_MERGE_INDEX_IN_SEGMENT, "false")
@@ -396,6 +439,9 @@ class StandardPartitionTableLoadingTestCase extends QueryTest with BeforeAndAfte
override def afterAll = {
dropTable
+ if (executorService != null && !executorService.isShutdown) {
+ executorService.shutdownNow()
+ }
}
def dropTable = {
@@ -413,6 +459,7 @@ class StandardPartitionTableLoadingTestCase extends QueryTest with BeforeAndAfte
sql("drop table if exists streamingpartitionedtable")
sql("drop table if exists mergeindexpartitionthree")
sql("drop table if exists loadstaticpartitiononeissue")
+ sql("drop table if exists partitionmultiplethreeconcurrent")
sql("drop table if exists loadpartitionwithspecialchar")
sql("drop table if exists emp1")
sql("drop table if exists restorepartition")
http://git-wip-us.apache.org/repos/asf/carbondata/blob/af4277e9/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/standardpartition/StandardPartitionTableOverwriteTestCase.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/standardpartition/StandardPartitionTableOverwriteTestCase.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/standardpartition/StandardPartitionTableOverwriteTestCase.scala
index 945542a..15126b6 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/standardpartition/StandardPartitionTableOverwriteTestCase.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/standardpartition/StandardPartitionTableOverwriteTestCase.scala
@@ -16,6 +16,7 @@
*/
package org.apache.carbondata.spark.testsuite.standardpartition
+import org.apache.spark.sql.Row
import org.apache.spark.sql.test.util.QueryTest
import org.scalatest.BeforeAndAfterAll
@@ -85,6 +86,75 @@ class StandardPartitionTableOverwriteTestCase extends QueryTest with BeforeAndAf
sql("select empno, empname,designation,workgroupcategory,workgroupcategoryname,deptno,projectjoindate,attendance,deptname,projectcode,utilization,salary,projectenddate,doj from originTable where projectenddate=cast('2016-06-29' as Date)"))
}
+ test("dynamic and static partition table with load syntax") {
+ sql(
+ """
+ | CREATE TABLE loadstaticpartitiondynamic (designation String, doj Timestamp,
+ | workgroupcategory int, workgroupcategoryname String, deptno int, deptname String,
+ | projectcode int, projectjoindate Timestamp, projectenddate Timestamp,attendance int,
+ | utilization int,salary int)
+ | PARTITIONED BY (empno int, empname String)
+ | STORED BY 'org.apache.carbondata.format'
+ """.stripMargin)
+
+ sql(s"""LOAD DATA local inpath '$resourcesPath/data.csv' INTO TABLE loadstaticpartitiondynamic PARTITION(empno='1', empname) OPTIONS('DELIMITER'= ',', 'QUOTECHAR'= '"')""")
+ checkAnswer(sql(s"select count(*) from loadstaticpartitiondynamic where empno=1"), sql(s"select count(*) from loadstaticpartitiondynamic"))
+ }
+
+ test("dynamic and static partition table with overwrite ") {
+ sql(
+ """
+ | CREATE TABLE insertstaticpartitiondynamic (designation String, doj Timestamp,salary int)
+ | PARTITIONED BY (empno int, empname String)
+ | STORED BY 'org.apache.carbondata.format'
+ """.stripMargin)
+ sql(s"""LOAD DATA local inpath '$resourcesPath/data.csv' INTO TABLE insertstaticpartitiondynamic PARTITION(empno, empname) OPTIONS('DELIMITER'= ',', 'QUOTECHAR'= '"')""")
+ val rows = sql(s"select count(*) from insertstaticpartitiondynamic").collect()
+ sql("""insert overwrite table insertstaticpartitiondynamic PARTITION(empno='1', empname) select designation, doj, salary, empname from insertstaticpartitiondynamic""")
+
+ checkAnswer(sql(s"select count(*) from insertstaticpartitiondynamic where empno=1"), rows)
+
+ intercept[Exception] {
+ sql("""insert overwrite table insertstaticpartitiondynamic PARTITION(empno, empname='ravi') select designation, doj, salary, empname from insertstaticpartitiondynamic""")
+ }
+
+ }
+
+ test("overwriting all partition on table and do compaction") {
+ sql(
+ """
+ | CREATE TABLE partitionallcompaction (empno int, empname String, designation String,
+ | workgroupcategory int, workgroupcategoryname String, deptno int,
+ | projectjoindate Timestamp, projectenddate Date,attendance int,
+ | utilization int,salary int)
+ | PARTITIONED BY (deptname String,doj Timestamp,projectcode int)
+ | STORED BY 'org.apache.carbondata.format'
+ """.stripMargin)
+ sql(s"""LOAD DATA local inpath '$resourcesPath/data.csv' OVERWRITE INTO TABLE partitionallcompaction OPTIONS('DELIMITER'= ',', 'QUOTECHAR'= '"')""")
+ sql(s"""LOAD DATA local inpath '$resourcesPath/data.csv' OVERWRITE INTO TABLE partitionallcompaction PARTITION(deptname='Learning', doj, projectcode) OPTIONS('DELIMITER'= ',', 'QUOTECHAR'= '"') """)
+ sql(s"""LOAD DATA local inpath '$resourcesPath/data.csv' OVERWRITE INTO TABLE partitionallcompaction PARTITION(deptname='configManagement', doj, projectcode) OPTIONS('DELIMITER'= ',', 'QUOTECHAR'= '"')""")
+ sql(s"""LOAD DATA local inpath '$resourcesPath/data.csv' OVERWRITE INTO TABLE partitionallcompaction PARTITION(deptname='network', doj, projectcode) OPTIONS('DELIMITER'= ',', 'QUOTECHAR'= '"')""")
+ sql(s"""LOAD DATA local inpath '$resourcesPath/data.csv' OVERWRITE INTO TABLE partitionallcompaction PARTITION(deptname='protocol', doj, projectcode) OPTIONS('DELIMITER'= ',', 'QUOTECHAR'= '"')""")
+ sql(s"""LOAD DATA local inpath '$resourcesPath/data.csv' OVERWRITE INTO TABLE partitionallcompaction PARTITION(deptname='security', doj, projectcode) OPTIONS('DELIMITER'= ',', 'QUOTECHAR'= '"')""")
+ sql("ALTER TABLE partitionallcompaction COMPACT 'MAJOR'").collect()
+ checkExistence(sql(s"""SHOW segments for table partitionallcompaction"""), true, "Marked for Delete")
+ }
+
+ test("Test overwrite static partition ") {
+ sql(
+ """
+ | CREATE TABLE weather6 (type String)
+ | PARTITIONED BY (year int, month int, day int)
+ | STORED BY 'org.apache.carbondata.format'
+ """.stripMargin)
+
+ sql("insert into weather6 partition(year=2014, month=5, day=25) select 'rainy'")
+ sql("insert into weather6 partition(year=2014, month=4, day=23) select 'cloudy'")
+ sql("insert overwrite table weather6 partition(year=2014, month=5, day=25) select 'sunny'")
+ checkExistence(sql("select * from weather6"), true, "sunny")
+ checkAnswer(sql("select count(*) from weather6"), Seq(Row(2)))
+ }
+
override def afterAll = {
dropTable
@@ -94,6 +164,10 @@ class StandardPartitionTableOverwriteTestCase extends QueryTest with BeforeAndAf
sql("drop table if exists originTable")
sql("drop table if exists partitiondateinsert")
sql("drop table if exists staticpartitiondateinsert")
+ sql("drop table if exists loadstaticpartitiondynamic")
+ sql("drop table if exists insertstaticpartitiondynamic")
+ sql("drop table if exists partitionallcompaction")
+ sql("drop table if exists weather6")
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/af4277e9/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDropPartitionRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDropPartitionRDD.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDropPartitionRDD.scala
index d377c4d..0a79295 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDropPartitionRDD.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDropPartitionRDD.scala
@@ -37,13 +37,16 @@ case class CarbonDropPartition(rddId: Int, val idx: Int, segmentPath: String)
* @param sc
* @param tablePath
* @param segments segments to be merged
+ * @param partialMatch If it is true then even the partial partition spec matches also can be
+ * dropped
*/
class CarbonDropPartitionRDD(
sc: SparkContext,
tablePath: String,
segments: Seq[String],
partitions: Seq[String],
- uniqueId: String)
+ uniqueId: String,
+ partialMatch: Boolean)
extends CarbonRDD[String](sc, Nil) {
override def getPartitions: Array[Partition] = {
@@ -60,7 +63,8 @@ class CarbonDropPartitionRDD(
new PartitionMapFileStore().dropPartitions(
split.segmentPath,
partitions.toList.asJava,
- uniqueId)
+ uniqueId,
+ partialMatch)
var havePair = false
var finished = false
http://git-wip-us.apache.org/repos/asf/carbondata/blob/af4277e9/integration/spark-common/src/main/scala/org/apache/spark/sql/catalyst/CarbonDDLSqlParser.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/spark/sql/catalyst/CarbonDDLSqlParser.scala b/integration/spark-common/src/main/scala/org/apache/spark/sql/catalyst/CarbonDDLSqlParser.scala
index 66351e3..3da603b 100644
--- a/integration/spark-common/src/main/scala/org/apache/spark/sql/catalyst/CarbonDDLSqlParser.scala
+++ b/integration/spark-common/src/main/scala/org/apache/spark/sql/catalyst/CarbonDDLSqlParser.scala
@@ -998,10 +998,10 @@ abstract class CarbonDDLSqlParser extends AbstractCarbonSparkSQLParser {
case _ => ("", "")
}
- protected lazy val partitions: Parser[(String, String)] =
- (ident <~ "=") ~ stringLit ^^ {
+ protected lazy val partitions: Parser[(String, Option[String])] =
+ (ident <~ "=".?) ~ stringLit.? ^^ {
case opt ~ optvalue => (opt.trim, optvalue)
- case _ => ("", "")
+ case _ => ("", None)
}
protected lazy val valueOptions: Parser[(Int, Int)] =
http://git-wip-us.apache.org/repos/asf/carbondata/blob/af4277e9/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala
index 0c6879c..60adcd7 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala
@@ -602,7 +602,7 @@ case class CarbonLoadDataCommand(
val lowerCasePartition = partition.map{case(key, value) => (key.toLowerCase, value)}
catalogTable.schema.map { attr =>
attributes.find(_.name.equalsIgnoreCase(attr.name)).get
- }.filter(attr => lowerCasePartition.get(attr.name.toLowerCase).isEmpty)
+ }.filter(attr => lowerCasePartition.getOrElse(attr.name.toLowerCase, None).isEmpty)
} else {
catalogTable.schema.map(f => attributes.find(_.name.equalsIgnoreCase(f.name)).get)
}
@@ -713,7 +713,12 @@ case class CarbonLoadDataCommand(
options += (("onepass", loadModel.getUseOnePass.toString))
options += (("dicthost", loadModel.getDictionaryServerHost))
options += (("dictport", loadModel.getDictionaryServerPort.toString))
- options += (("staticpartition", partition.nonEmpty.toString))
+ if (partition.nonEmpty) {
+ val staticPartitionStr = ObjectSerializationUtil.convertObjectToString(
+ new util.HashMap[String, Boolean](
+ partition.map{case (col, value) => (col.toLowerCase, value.isDefined)}.asJava))
+ options += (("staticpartition", staticPartitionStr))
+ }
options += (("operationcontext", operationContextStr))
options ++= this.options
if (updateModel.isDefined) {
@@ -745,63 +750,80 @@ case class CarbonLoadDataCommand(
sparkSession: SparkSession,
table: CarbonTable,
logicalPlan: LogicalPlan): Unit = {
- sparkSession.sessionState.catalog.listPartitions(
- TableIdentifier(table.getTableName, Some(table.getDatabaseName)),
- Some(partition.map(f => (f._1, f._2.get))))
- val partitionNames = partition.map(k => k._1 + "=" + k._2.get).toSet
+ val identifier = TableIdentifier(table.getTableName, Some(table.getDatabaseName))
+ val existingPartitions = sparkSession.sessionState.catalog.listPartitions(
+ identifier,
+ Some(partition.filter(_._2.isDefined).map(f => (f._1, f._2.get))))
+ val partitionNames = existingPartitions.toList.flatMap { partition =>
+ partition.spec.seq.map{case (column, value) => column + "=" + value}
+ }.toSet
val uniqueId = System.currentTimeMillis().toString
val segments = new SegmentStatusManager(
table.getAbsoluteTableIdentifier).getValidAndInvalidSegments.getValidSegments
- try {
- // First drop the partitions from partition mapper files of each segment
- new CarbonDropPartitionRDD(
- sparkSession.sparkContext,
- table.getTablePath,
- segments.asScala,
- partitionNames.toSeq,
- uniqueId).collect()
- } catch {
- case e: Exception =>
- // roll back the drop partitions from carbon store
- new CarbonDropPartitionCommitRDD(
+ // If any existing partitions need to be overwritten then drop from partitionmap
+ if (partitionNames.nonEmpty) {
+ try {
+ // First drop the partitions from partition mapper files of each segment
+ new CarbonDropPartitionRDD(
sparkSession.sparkContext,
table.getTablePath,
segments.asScala,
- false,
- uniqueId).collect()
- throw e
- }
+ partitionNames.toSeq,
+ uniqueId,
+ partialMatch = false).collect()
+ } catch {
+ case e: Exception =>
+ // roll back the drop partitions from carbon store
+ new CarbonDropPartitionCommitRDD(
+ sparkSession.sparkContext,
+ table.getTablePath,
+ segments.asScala,
+ success = false,
+ uniqueId).collect()
+ throw e
+ }
- try {
+ try {
+ Dataset.ofRows(sparkSession, logicalPlan)
+ } catch {
+ case e: Exception =>
+ // roll back the drop partitions from carbon store
+ new CarbonDropPartitionCommitRDD(
+ sparkSession.sparkContext,
+ table.getTablePath,
+ segments.asScala,
+ success = false,
+ uniqueId).collect()
+ throw e
+ }
+ // Commit the removed partitions in carbon store.
+ new CarbonDropPartitionCommitRDD(
+ sparkSession.sparkContext,
+ table.getTablePath,
+ segments.asScala,
+ success = true,
+ uniqueId).collect()
+ // get valid segments
+ val validsegments =
+ new SegmentStatusManager(
+ table.getAbsoluteTableIdentifier).getValidAndInvalidSegments.getValidSegments
+ // Update the loadstatus with update time to clear cache from driver.
+ CarbonUpdateUtil.updateTableMetadataStatus(
+ new util.HashSet[String](validsegments),
+ table,
+ uniqueId,
+ true,
+ new util.ArrayList[String])
+ DataMapStoreManager.getInstance().clearDataMaps(table.getAbsoluteTableIdentifier)
+ // Clean the overwriting segments if any.
+ new PartitionMapFileStore().cleanSegments(
+ table,
+ CarbonFilters.getPartitions(Seq.empty, sparkSession, identifier).asJava,
+ false)
+ } else {
+ // Otherwise its a normal load
Dataset.ofRows(sparkSession, logicalPlan)
- } catch {
- case e: Exception =>
- // roll back the drop partitions from carbon store
- new CarbonDropPartitionCommitRDD(
- sparkSession.sparkContext,
- table.getTablePath,
- segments.asScala,
- false,
- uniqueId).collect()
- throw e
}
- // Commit the removed partitions in carbon store.
- new CarbonDropPartitionCommitRDD(
- sparkSession.sparkContext,
- table.getTablePath,
- segments.asScala,
- true,
- uniqueId).collect()
- // Update the loadstatus with update time to clear cache from driver.
- val segmentSet = new util.HashSet[String](new SegmentStatusManager(table
- .getAbsoluteTableIdentifier).getValidAndInvalidSegments.getValidSegments)
- CarbonUpdateUtil.updateTableMetadataStatus(
- segmentSet,
- table,
- uniqueId,
- true,
- new util.ArrayList[String])
- DataMapStoreManager.getInstance().clearDataMaps(table.getAbsoluteTableIdentifier)
}
def getDataFrameWithTupleID(): DataFrame = {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/af4277e9/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/partition/CarbonAlterTableDropHivePartitionCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/partition/CarbonAlterTableDropHivePartitionCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/partition/CarbonAlterTableDropHivePartitionCommand.scala
index c68e43c..dbd686b 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/partition/CarbonAlterTableDropHivePartitionCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/partition/CarbonAlterTableDropHivePartitionCommand.scala
@@ -25,11 +25,13 @@ import org.apache.spark.sql.{CarbonEnv, Row, SparkSession}
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
import org.apache.spark.sql.execution.command.{AlterTableAddPartitionCommand, AlterTableDropPartitionCommand, AtomicRunnableCommand}
+import org.apache.spark.sql.optimizer.CarbonFilters
import org.apache.spark.util.AlterTableUtil
import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.datamap.DataMapStoreManager
import org.apache.carbondata.core.locks.{ICarbonLock, LockUsage}
+import org.apache.carbondata.core.metadata.PartitionMapFileStore
import org.apache.carbondata.core.mutate.CarbonUpdateUtil
import org.apache.carbondata.core.statusmanager.SegmentStatusManager
import org.apache.carbondata.spark.rdd.{CarbonDropPartitionCommitRDD, CarbonDropPartitionRDD}
@@ -73,15 +75,19 @@ case class CarbonAlterTableDropHivePartitionCommand(
}
// Drop the partitions from hive.
- AlterTableDropPartitionCommand(tableName, specs, ifExists, purge, retainData)
- .run(sparkSession)
+ AlterTableDropPartitionCommand(
+ tableName,
+ specs,
+ ifExists,
+ purge,
+ retainData).run(sparkSession)
}
Seq.empty[Row]
}
override def undoMetadata(sparkSession: SparkSession, exception: Exception): Seq[Row] = {
- AlterTableAddPartitionCommand(tableName, specs.map((_, None)), ifExists)
+ AlterTableAddPartitionCommand(tableName, specs.map((_, None)), true)
val msg = s"Got exception $exception when processing data of drop partition." +
"Adding back partitions to the metadata"
LogServiceFactory.getLogService(this.getClass.getCanonicalName).error(msg)
@@ -114,7 +120,8 @@ case class CarbonAlterTableDropHivePartitionCommand(
table.getTablePath,
segments.asScala,
partitionNames.toSeq,
- uniqueId).collect()
+ uniqueId,
+ partialMatch = true).collect()
} catch {
case e: Exception =>
// roll back the drop partitions from carbon store
@@ -143,6 +150,10 @@ case class CarbonAlterTableDropHivePartitionCommand(
DataMapStoreManager.getInstance().clearDataMaps(table.getAbsoluteTableIdentifier)
} finally {
AlterTableUtil.releaseLocks(locks)
+ new PartitionMapFileStore().cleanSegments(
+ table,
+ new util.ArrayList(CarbonFilters.getPartitions(Seq.empty, sparkSession, tableName).asJava),
+ false)
}
Seq.empty[Row]
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/af4277e9/integration/spark2/src/main/scala/org/apache/spark/sql/execution/datasources/CarbonFileFormat.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/datasources/CarbonFileFormat.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/datasources/CarbonFileFormat.scala
index 4b368de..36df787 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/datasources/CarbonFileFormat.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/datasources/CarbonFileFormat.scala
@@ -42,6 +42,7 @@ import org.apache.carbondata.core.util.CarbonProperties
import org.apache.carbondata.core.util.path.CarbonTablePath
import org.apache.carbondata.hadoop.api.{CarbonOutputCommitter, CarbonTableOutputFormat}
import org.apache.carbondata.hadoop.api.CarbonTableOutputFormat.CarbonRecordWriter
+import org.apache.carbondata.hadoop.util.ObjectSerializationUtil
import org.apache.carbondata.processing.loading.csvinput.StringArrayWritable
import org.apache.carbondata.processing.loading.model.CarbonLoadModel
import org.apache.carbondata.spark.util.{CarbonScalaUtil, DataLoadingUtil, Util}
@@ -204,9 +205,14 @@ private class CarbonOutputWriter(path: String,
fieldTypes: Seq[DataType])
extends OutputWriter with AbstractCarbonOutputWriter {
val partitions = getPartitionsFromPath(path, context).map(ExternalCatalogUtils.unescapePathName)
- val staticPartition = {
+ val staticPartition: util.HashMap[String, Boolean] = {
val staticPart = context.getConfiguration.get("carbon.staticpartition")
- staticPart != null && staticPart.toBoolean
+ if (staticPart != null) {
+ ObjectSerializationUtil.convertStringToObject(
+ staticPart).asInstanceOf[util.HashMap[String, Boolean]]
+ } else {
+ null
+ }
}
lazy val partitionData = if (partitions.nonEmpty) {
val updatedPartitions = partitions.map{ p =>
@@ -223,7 +229,7 @@ private class CarbonOutputWriter(path: String,
}
}
- if (staticPartition) {
+ if (staticPartition != null) {
val loadModel = recordWriter.getLoadModel
val table = loadModel.getCarbonDataLoadSchema.getCarbonTable
var timeStampformatString = loadModel.getTimestampformat
@@ -237,11 +243,17 @@ private class CarbonOutputWriter(path: String,
}
val dateFormat = new SimpleDateFormat(dateFormatString)
updatedPartitions.map {case (col, value) =>
- CarbonScalaUtil.convertToCarbonFormat(value,
- CarbonScalaUtil.convertCarbonToSparkDataType(
- table.getColumnByName(table.getTableName, col).getDataType),
- timeFormat,
- dateFormat)
+ // Only convert the static partitions to the carbon format and use it while loading data
+ // to carbon.
+ if (staticPartition.getOrDefault(col, false)) {
+ CarbonScalaUtil.convertToCarbonFormat(value,
+ CarbonScalaUtil.convertCarbonToSparkDataType(
+ table.getColumnByName(table.getTableName, col).getDataType),
+ timeFormat,
+ dateFormat)
+ } else {
+ value
+ }
}
} else {
updatedPartitions.map(_._2)
@@ -309,9 +321,25 @@ private class CarbonOutputWriter(path: String,
(col, value)
}.toMap
val updatedPartitions =
- if (staticPartition) {
- splitPartitions
+ if (staticPartition != null) {
+ // There can be scnerio like dynamic and static combination, in that case we should convert
+ // only the dyanamic partition values to the proper format and store to carbon parttion map
+ splitPartitions.map { case (col, value) =>
+ if (!staticPartition.getOrDefault(col, false)) {
+ CarbonScalaUtil.updatePartitions(
+ Seq((col, value)).toMap,
+ table,
+ timeFormat,
+ dateFormat,
+ serializeFormat,
+ badRecordAction,
+ isEmptyBadRecord).toSeq.head
+ } else {
+ (col, value)
+ }
+ }
} else {
+ // All dynamic partitions need to be converted to proper format
CarbonScalaUtil.updatePartitions(
splitPartitions,
table,
http://git-wip-us.apache.org/repos/asf/carbondata/blob/af4277e9/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala
index a25be06..4045478 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala
@@ -361,7 +361,7 @@ class CarbonSpark2SqlParser extends CarbonDDLSqlParser {
validateOptions(optionsList)
}
val optionsMap = optionsList.getOrElse(List.empty[(String, String)]).toMap
- val partitionSpec = partitions.getOrElse(List.empty[(String, String)]).toMap
+ val partitionSpec = partitions.getOrElse(List.empty[(String, Option[String])]).toMap
CarbonLoadDataCommand(
databaseNameOp = convertDbNameToLowerCase(databaseNameOp),
tableName = tableName,
@@ -374,7 +374,7 @@ class CarbonSpark2SqlParser extends CarbonDDLSqlParser {
updateModel = None,
tableInfoOp = None,
internalOptions = Map.empty,
- partition = partitionSpec.map { case (key, value) => (key, Some(value))})
+ partition = partitionSpec)
}
protected lazy val deleteLoadsByID: Parser[LogicalPlan] =
[5/7] carbondata git commit: [CARBONDATA-1754][BugFix] Fixed issue
occuring on concurrent insert-overwrite and compaction
Posted by ja...@apache.org.
[CARBONDATA-1754][BugFix] Fixed issue occuring on concurrent insert-overwrite and compaction
Description: Concurrent Insert overwrite-Compaction: Compaction job fails at run time if insert overwrite job is running concurrently.
Problem: When Insert-overwrite and compaction are run from two different sessions, Insert overwrite job is success but after that compaction fails with an exception saying 'Compaction failed to update metadata for table'
Solution: Ideally compaction job should give an exception in start with message that 'insert overwrite is in progress'.
This closes #1711
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/11353e2d
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/11353e2d
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/11353e2d
Branch: refs/heads/carbonstore
Commit: 11353e2d0475689a1220ee9022878c655e79f298
Parents: 705b111
Author: SangeetaGulia <sa...@knoldus.in>
Authored: Thu Dec 21 18:42:18 2017 +0530
Committer: Jacky Li <ja...@qq.com>
Committed: Mon Jan 8 16:30:54 2018 +0800
----------------------------------------------------------------------
.../iud/TestInsertOverwriteAndCompaction.scala | 104 +++++++++++++++++++
.../exception/ConcurrentOperationException.java | 44 ++++++++
.../CarbonAlterTableCompactionCommand.scala | 14 ++-
3 files changed, 160 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/11353e2d/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/TestInsertOverwriteAndCompaction.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/TestInsertOverwriteAndCompaction.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/TestInsertOverwriteAndCompaction.scala
new file mode 100644
index 0000000..f69d1e7
--- /dev/null
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/TestInsertOverwriteAndCompaction.scala
@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.carbondata.spark.testsuite.iud
+
+import java.text.SimpleDateFormat
+import java.util
+import java.util.concurrent.{Callable, ExecutorService, Executors, Future}
+
+import org.apache.spark.sql.test.util.QueryTest
+import org.apache.spark.sql.{DataFrame, SaveMode}
+import org.scalatest.BeforeAndAfterAll
+
+import org.apache.carbondata.core.constants.CarbonCommonConstants
+import org.apache.carbondata.core.util.CarbonProperties
+
+class TestInsertOverwriteAndCompaction extends QueryTest with BeforeAndAfterAll {
+ var df: DataFrame = _
+ private val executorService: ExecutorService = Executors.newFixedThreadPool(10)
+
+ override def beforeAll {
+ dropTable()
+ buildTestData()
+ }
+
+ override def afterAll {
+ executorService.shutdownNow()
+ dropTable()
+ }
+
+
+ private def buildTestData(): Unit = {
+
+ CarbonProperties.getInstance()
+ .addProperty(CarbonCommonConstants.CARBON_DATE_FORMAT, "yyyy-MM-dd")
+
+ // Simulate data and write to table orders
+ import sqlContext.implicits._
+
+ val sdf = new SimpleDateFormat("yyyy-MM-dd")
+ df = sqlContext.sparkSession.sparkContext.parallelize(1 to 1500000)
+ .map(value => (value, new java.sql.Date(sdf.parse("2015-07-" + (value % 10 + 10)).getTime),
+ "china", "aaa" + value, "phone" + 555 * value, "ASD" + (60000 + value), 14999 + value,"ordersTable"+value))
+ .toDF("o_id", "o_date", "o_country", "o_name",
+ "o_phonetype", "o_serialname", "o_salary","o_comment")
+ createTable("orders")
+ createTable("orders_overwrite")
+ }
+
+ private def dropTable() = {
+ sql("DROP TABLE IF EXISTS orders")
+ sql("DROP TABLE IF EXISTS orders_overwrite")
+ }
+
+ private def createTable(tableName: String): Unit ={
+ df.write
+ .format("carbondata")
+ .option("tableName", tableName)
+ .option("tempCSV", "true")
+ .option("compress", "true")
+ .mode(SaveMode.Overwrite)
+ .save()
+ }
+
+ test("Concurrency test for Insert-Overwrite and compact") {
+ val tasks = new java.util.ArrayList[Callable[String]]()
+ tasks.add(new QueryTask(s"insert overWrite table orders select * from orders_overwrite"))
+ tasks.add(new QueryTask("alter table orders compact 'MINOR'"))
+ val results: util.List[Future[String]] = executorService.invokeAll(tasks)
+ val resultList = new util.ArrayList[String]()
+ resultList.add(results.get(0).get)
+ resultList.add(results.get(1).get)
+ assert(resultList.contains("PASS"))
+ }
+
+ class QueryTask(query: String) extends Callable[String] {
+ override def call(): String = {
+ var result = "PASS"
+ try {
+ LOGGER.info("Executing :" + query + Thread.currentThread().getName)
+ sql(query).show()
+ } catch {
+ case _: Exception =>
+ result = "FAIL"
+ }
+ result
+ }
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/carbondata/blob/11353e2d/integration/spark-common/src/main/java/org/apache/carbondata/spark/exception/ConcurrentOperationException.java
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/java/org/apache/carbondata/spark/exception/ConcurrentOperationException.java b/integration/spark-common/src/main/java/org/apache/carbondata/spark/exception/ConcurrentOperationException.java
new file mode 100644
index 0000000..1f3c07d
--- /dev/null
+++ b/integration/spark-common/src/main/java/org/apache/carbondata/spark/exception/ConcurrentOperationException.java
@@ -0,0 +1,44 @@
+/*
+ * 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.exception;
+
+public class ConcurrentOperationException extends Exception {
+
+ /**
+ * The Error message.
+ */
+ private String msg = "";
+
+ /**
+ * Constructor
+ *
+ * @param msg The error message for this exception.
+ */
+ public ConcurrentOperationException(String msg) {
+ super(msg);
+ this.msg = msg;
+ }
+
+ /**
+ * getMessage
+ */
+ public String getMessage() {
+ return this.msg;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/11353e2d/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonAlterTableCompactionCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonAlterTableCompactionCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonAlterTableCompactionCommand.scala
index 6daaae5..b10777d 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonAlterTableCompactionCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonAlterTableCompactionCommand.scala
@@ -36,11 +36,13 @@ import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.locks.{CarbonLockFactory, LockUsage}
import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, TableInfo}
import org.apache.carbondata.core.mutate.CarbonUpdateUtil
+import org.apache.carbondata.core.statusmanager.SegmentStatusManager
import org.apache.carbondata.core.util.CarbonProperties
import org.apache.carbondata.core.util.path.CarbonStorePath
import org.apache.carbondata.events.{AlterTableCompactionPostEvent, AlterTableCompactionPreEvent, AlterTableCompactionPreStatusUpdateEvent, OperationContext, OperationListenerBus}
import org.apache.carbondata.processing.loading.model.{CarbonDataLoadSchema, CarbonLoadModel}
import org.apache.carbondata.processing.merger.{CarbonDataMergerUtil, CompactionType}
+import org.apache.carbondata.spark.exception.ConcurrentOperationException
import org.apache.carbondata.spark.rdd.CarbonDataRDDFactory
import org.apache.carbondata.spark.util.CommonUtil
import org.apache.carbondata.streaming.StreamHandoffRDD
@@ -56,7 +58,7 @@ case class CarbonAlterTableCompactionCommand(
override def processData(sparkSession: SparkSession): Seq[Row] = {
val LOGGER: LogService =
- LogServiceFactory.getLogService(this.getClass.getName)
+ LogServiceFactory.getLogService(this.getClass.getName)
val tableName = alterTableModel.tableName.toLowerCase
val databaseName = alterTableModel.dbName.getOrElse(sparkSession.catalog.currentDatabase)
@@ -79,6 +81,14 @@ case class CarbonAlterTableCompactionCommand(
relation.carbonTable
}
+ val isLoadInProgress = SegmentStatusManager.checkIfAnyLoadInProgressForTable(table)
+ if (isLoadInProgress) {
+ val message = "Cannot run data loading and compaction on same table concurrently. " +
+ "Please wait for load to finish"
+ LOGGER.error(message)
+ throw new ConcurrentOperationException(message)
+ }
+
val carbonLoadModel = new CarbonLoadModel()
carbonLoadModel.setTableName(table.getTableName)
val dataLoadSchema = new CarbonDataLoadSchema(table)
@@ -162,7 +172,7 @@ case class CarbonAlterTableCompactionCommand(
}
// reading the start time of data load.
- val loadStartTime : Long =
+ val loadStartTime: Long =
if (alterTableModel.factTimeStamp.isEmpty) {
CarbonUpdateUtil.readCurrentTime
} else {
[2/7] carbondata git commit:
[CARBONDATA-1926][CARBONDATA-1927][Pre-Aggregate] Expression support inside
aggregate function for Query
Posted by ja...@apache.org.
http://git-wip-us.apache.org/repos/asf/carbondata/blob/c70e73f1/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonPreAggregateRules.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonPreAggregateRules.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonPreAggregateRules.scala
index 1c93617..bd8b4c6 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonPreAggregateRules.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonPreAggregateRules.scala
@@ -18,29 +18,46 @@
package org.apache.spark.sql.hive
import scala.collection.JavaConverters._
-import scala.collection.mutable.ArrayBuffer
+import scala.collection.mutable
-import org.apache.spark.SPARK_VERSION
import org.apache.spark.sql._
-import org.apache.spark.sql.CarbonExpressions.{CarbonScalaUDF, CarbonSubqueryAlias, MatchCast, MatchCastExpression}
import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.CarbonExpressions.{CarbonScalaUDF, CarbonSubqueryAlias, MatchCastExpression}
import org.apache.spark.sql.catalyst.analysis.{UnresolvedAlias, UnresolvedAttribute}
-import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeReference, Cast, Divide, Expression, Literal, NamedExpression, ScalaUDF, SortOrder}
-import org.apache.spark.sql.catalyst.expressions.aggregate._
+import org.apache.spark.sql.catalyst.expressions.{Alias, AttributeReference, Cast, Divide, Expression, Literal, NamedExpression, ScalaUDF, SortOrder}
+import org.apache.spark.sql.catalyst.expressions.aggregate.{Count, _}
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.execution.command.preaaggregate.PreAggregateUtil
import org.apache.spark.sql.execution.datasources.{FindDataSourceTable, LogicalRelation}
+import org.apache.spark.sql.parser.CarbonSpark2SqlParser
import org.apache.spark.sql.types._
-import org.apache.spark.sql.util.CarbonException
import org.apache.spark.util.CarbonReflectionUtils
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.metadata.schema.table.{AggregationDataMapSchema, CarbonTable, DataMapSchema}
-import org.apache.carbondata.core.preagg.{AggregateTableSelector, QueryColumn, QueryPlan}
+import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema
+import org.apache.carbondata.core.preagg.{AggregateQueryPlan, AggregateTableSelector, QueryColumn}
import org.apache.carbondata.core.util.{CarbonUtil, ThreadLocalSessionInfo}
-import org.apache.carbondata.spark.util.CarbonScalaUtil
/**
+ * model class to store aggregate expression logical plan
+ * and its column schema mapping
+ * @param expression aggregate expression
+ * @param columnSchema list of column schema from table
+ */
+case class AggExpToColumnMappingModel(
+ expression: Expression,
+ var columnSchema: Option[Object] = None) {
+ override def equals(o: Any) : Boolean = o match {
+ case that: AggExpToColumnMappingModel =>
+ that.expression==this.expression
+ case _ => false
+ }
+ // TODO need to update the hash code generation code
+ override def hashCode : Int = 1
+}
+/**
* Class for applying Pre Aggregate rules
* Responsibility.
* 1. Check plan is valid plan for updating the parent table plan with child table
@@ -73,11 +90,36 @@ import org.apache.carbondata.spark.util.CarbonScalaUtil
* 6.1 validate maintable has timeseries datamap
* 6.2 timeseries function is valid function or not
*
- * @param sparkSession
- * spark session
+ * @param sparkSession spark session
*/
case class CarbonPreAggregateQueryRules(sparkSession: SparkSession) extends Rule[LogicalPlan] {
+ /**
+ * map for keeping parent attribute reference to child attribute reference
+ * this will be used to updated the plan in case of join or order by
+ */
+ val updatedExpression = mutable.HashMap[AttributeReference, AttributeReference]()
+
+ /**
+ * parser
+ */
+ lazy val parser = new CarbonSpark2SqlParser
+
+ /**
+ * Below method will be used to validate the logical plan
+ * @param logicalPlan query logical plan
+ * @return isvalid or not
+ */
+ private def isValidPlan(logicalPlan: LogicalPlan) : Boolean = {
+ var isValidPlan = true
+ logicalPlan.transform {
+ case aggregate@Aggregate(grp, aExp, child) =>
+ isValidPlan = !aExp.exists(p => p.name.equals("preAggLoad") || p.name.equals("preAgg"))
+ val updatedAggExp = aExp.filterNot(_.name.equalsIgnoreCase("preAggLoad"))
+ Aggregate(grp, updatedAggExp, child)
+ }
+ isValidPlan
+ }
override def apply(plan: LogicalPlan): LogicalPlan = {
var needAnalysis = true
plan.transformExpressions {
@@ -98,244 +140,403 @@ case class CarbonPreAggregateQueryRules(sparkSession: SparkSession) extends Rule
needAnalysis = false
attr
}
+ if(needAnalysis) {
+ needAnalysis = isValidPlan(plan)
+ }
// if plan is not valid for transformation then return same plan
if (!needAnalysis) {
plan
} else {
- // create buffer to collect all the column and its metadata information
- val list = scala.collection.mutable.HashSet.empty[QueryColumn]
- var isValidPlan = true
- val carbonTable = plan match {
- // matching the plan based on supported plan
- // if plan is matches with any case it will validate and get all
- // information required for transforming the plan
+ val updatedPlan = transformPreAggQueryPlan(plan)
+ val newPlan = updatePlan(updatedPlan)
+ newPlan
+ }
+ }
- // When plan has grouping expression, aggregate expression
- // subquery
- case Aggregate(groupingExp,
- aggregateExp,
- CarbonSubqueryAlias(_, logicalRelation: LogicalRelation))
- // only carbon query plan is supported checking whether logical relation is
- // is for carbon
- if logicalRelation.relation.isInstanceOf[CarbonDatasourceHadoopRelation] &&
- logicalRelation.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonRelation.
- metaData.hasAggregateDataMapSchema =>
- val (carbonTable, tableName) = getCarbonTableAndTableName(logicalRelation)
- // if it is valid plan then extract the query columns
- isValidPlan = extractQueryColumnsFromAggExpression(groupingExp,
- aggregateExp,
- carbonTable,
- tableName,
- list)
- carbonTable
+ /**
+ * Below method will be used to update the child plan
+ * This will be used for updating expression like join condition,
+ * order by, project list etc
+ * @param plan child plan
+ * @return updated plan
+ */
+ def updatePlan(plan: LogicalPlan) : LogicalPlan = {
+ val updatedPlan = plan transform {
+ case Aggregate(grp, aggExp, child) =>
+ Aggregate(
+ updateExpression(grp),
+ updateExpression(aggExp.asInstanceOf[Seq[Expression]]).asInstanceOf[Seq[NamedExpression]],
+ child)
+ case Filter(filterExp, child) =>
+ Filter(updateExpression(Seq(filterExp)).head, child)
+ case Project(pList, child) =>
+ Project(
+ updateExpression(pList.asInstanceOf[Seq[Expression]]).asInstanceOf[Seq[NamedExpression]],
+ child)
+ case Sort(sortOrders, global, child) =>
+ Sort(updateSortExpression(sortOrders), global, child)
+ case Join(left, right, joinType, condition) =>
+ val updatedCondition = condition match {
+ case Some(expr) => Some(updateExpression(Seq(expr)).head)
+ case _ => condition
+ }
+ Join(left, right, joinType, updatedCondition)
+ }
+ updatedPlan
+ }
- // below case for handling filter query
- // When plan has grouping expression, aggregate expression
- // filter expression
- case Aggregate(groupingExp,
- aggregateExp,
- Filter(filterExp,
- CarbonSubqueryAlias(_, logicalRelation: LogicalRelation)))
- // only carbon query plan is supported checking whether logical relation is
- // is for carbon
- if logicalRelation.relation.isInstanceOf[CarbonDatasourceHadoopRelation] &&
- logicalRelation.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonRelation.
- metaData.hasAggregateDataMapSchema =>
- val (carbonTable, tableName) = getCarbonTableAndTableName(logicalRelation)
- // if it is valid plan then extract the query columns
- isValidPlan = extractQueryColumnsFromAggExpression(groupingExp,
- aggregateExp,
- carbonTable,
- tableName,
- list)
- if(isValidPlan) {
- isValidPlan = !CarbonReflectionUtils.hasPredicateSubquery(filterExp)
- }
- // getting the columns from filter expression
- if(isValidPlan) {
- isValidPlan = extractQueryColumnFromFilterExp(filterExp, list, carbonTable, tableName)
+ /**
+ * Below method will be used to update the sort expression
+ * @param sortExp sort order expression in query
+ * @return updated sort expression
+ */
+ def updateSortExpression(sortExp : Seq[SortOrder]) : Seq[SortOrder] = {
+ sortExp map { order =>
+ SortOrder(order.child transform {
+ case attr: AttributeReference =>
+ updatedExpression.find { p => p._1.sameRef(attr) } match {
+ case Some((_, childAttr)) =>
+ AttributeReference(
+ childAttr.name,
+ childAttr.dataType,
+ childAttr.nullable,
+ childAttr.metadata)(childAttr.exprId, attr.qualifier, attr.isGenerated)
+ case None =>
+ attr
}
- carbonTable
+ }, order.direction )
+ }
+ }
- // When plan has grouping expression, aggregate expression
- // logical relation
- case Aggregate(groupingExp, aggregateExp, logicalRelation: LogicalRelation)
- // only carbon query plan is supported checking whether logical relation is
- // is for carbon
- if logicalRelation.relation.isInstanceOf[CarbonDatasourceHadoopRelation] &&
- logicalRelation.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonRelation.
- metaData.hasAggregateDataMapSchema =>
- val (carbonTable, tableName) = getCarbonTableAndTableName(logicalRelation)
- // if it is valid plan then extract the query columns
- isValidPlan = extractQueryColumnsFromAggExpression(groupingExp,
- aggregateExp,
- carbonTable,
- tableName,
- list)
- carbonTable
- // case for handling aggregation, order by
- case Project(projectList,
- Sort(sortOrders,
- _,
- Aggregate(groupingExp,
- aggregateExp,
- CarbonSubqueryAlias(_, logicalRelation: LogicalRelation))))
- if logicalRelation.relation.isInstanceOf[CarbonDatasourceHadoopRelation] &&
- logicalRelation.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonRelation.
- metaData.hasAggregateDataMapSchema =>
- val (carbonTable, tableName) = getCarbonTableAndTableName(logicalRelation)
- isValidPlan = extractQueryColumnsFromAggExpression(groupingExp,
- aggregateExp,
- carbonTable,
- tableName,
- list)
- if(isValidPlan) {
- list ++
- extractQueryColumnForOrderBy(Some(projectList), sortOrders, carbonTable, tableName)
- }
- carbonTable
- // case for handling aggregation, order by and filter
- case Project(projectList,
- Sort(sortOrders,
- _,
- Aggregate(groupingExp,
- aggregateExp,
- Filter(filterExp, CarbonSubqueryAlias(_, logicalRelation: LogicalRelation)))))
- if logicalRelation.relation.isInstanceOf[CarbonDatasourceHadoopRelation] &&
- logicalRelation.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonRelation.
- metaData.hasAggregateDataMapSchema =>
- val (carbonTable, tableName) = getCarbonTableAndTableName(logicalRelation)
- isValidPlan = extractQueryColumnsFromAggExpression(groupingExp,
- aggregateExp,
- carbonTable,
- tableName,
- list)
- if(isValidPlan) {
- isValidPlan = !CarbonReflectionUtils.hasPredicateSubquery(filterExp)
- }
- if (isValidPlan) {
- list ++
- extractQueryColumnForOrderBy(Some(projectList), sortOrders, carbonTable, tableName)
- isValidPlan = extractQueryColumnFromFilterExp(filterExp, list, carbonTable, tableName)
+ /**
+ * Below method will be used to update the expression like group by expression
+ * @param expressions sequence of expression like group by
+ * @return updated expressions
+ */
+ def updateExpression(expressions : Seq[Expression]) : Seq[Expression] = {
+ expressions map { expression =>
+ expression transform {
+ case attr: AttributeReference =>
+ updatedExpression.find { p => p._1.sameRef(attr) } match {
+ case Some((_, childAttr)) =>
+ AttributeReference(
+ childAttr.name,
+ childAttr.dataType,
+ childAttr.nullable,
+ childAttr.metadata)(childAttr.exprId, attr.qualifier, attr.isGenerated)
+ case None =>
+ attr
}
- carbonTable
- // case for handling aggregation with order by when only projection column exits
- case Sort(sortOrders,
- _,
- Aggregate(groupingExp,
- aggregateExp,
- CarbonSubqueryAlias(_, logicalRelation: LogicalRelation)))
- if logicalRelation.relation.isInstanceOf[CarbonDatasourceHadoopRelation] &&
- logicalRelation.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonRelation.
- metaData.hasAggregateDataMapSchema =>
- val (carbonTable, tableName) = getCarbonTableAndTableName(logicalRelation)
- isValidPlan = extractQueryColumnsFromAggExpression(groupingExp,
- aggregateExp,
+ }
+ }
+ }
+
+ /**
+ * Below method will be used to validate and transform the main table plan to child table plan
+ * rules for transforming is as below.
+ * 1. Grouping expression rules
+ * 1.1 Change the parent attribute reference for of group expression
+ * to child attribute reference
+ *
+ * 2. Aggregate expression rules
+ * 2.1 Change the parent attribute reference for of group expression to
+ * child attribute reference
+ * 2.2 Change the count AggregateExpression to Sum as count
+ * is already calculated so in case of aggregate table
+ * we need to apply sum to get the count
+ * 2.2 In case of average aggregate function select 2 columns from aggregate table with
+ * aggregation sum and count. Then add divide(sum(column with sum), sum(column with count)).
+ * Note: During aggregate table creation for average table will be created with two columns
+ * one for sum(column) and count(column) to support rollup
+ * 3. Filter Expression rules.
+ * 3.1 Updated filter expression attributes with child table attributes
+ * 4. Update the Parent Logical relation with child Logical relation
+ * 5. timeseries function
+ * 5.1 validate parent table has timeseries datamap
+ * 5.2 timeseries function is valid function or not
+ *
+ * @param logicalPlan parent logical plan
+ * @return transformed plan
+ */
+ def transformPreAggQueryPlan(logicalPlan: LogicalPlan): LogicalPlan = {
+ val updatedPlan = logicalPlan.transform {
+ // case for aggregation query
+ case agg@Aggregate(
+ grExp,
+ aggExp,
+ child@CarbonSubqueryAlias(_, l: LogicalRelation))
+ if l.relation.isInstanceOf[CarbonDatasourceHadoopRelation] &&
+ l.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonRelation.
+ metaData.hasAggregateDataMapSchema =>
+ val carbonTable = getCarbonTable(l)
+ val list = scala.collection.mutable.HashSet.empty[QueryColumn]
+ val aggregateExpressions = scala.collection.mutable.HashSet.empty[AggregateExpression]
+ val isValidPlan = extractQueryColumnsFromAggExpression(
+ grExp,
+ aggExp,
+ carbonTable,
+ list,
+ aggregateExpressions)
+ if(isValidPlan) {
+ val (aggDataMapSchema, childPlan) = getChildDataMapForTransformation(list,
+ aggregateExpressions,
carbonTable,
- tableName,
- list)
- if(isValidPlan) {
- list ++ extractQueryColumnForOrderBy(sortOrders = sortOrders,
- carbonTable = carbonTable,
- tableName = tableName)
+ agg)
+ if(null != aggDataMapSchema && null!= childPlan) {
+ val attributes = childPlan.output.asInstanceOf[Seq[AttributeReference]]
+ val (updatedGroupExp, updatedAggExp, newChild, None) =
+ getUpdatedExpressions(grExp,
+ aggExp,
+ child,
+ None,
+ aggDataMapSchema,
+ attributes,
+ childPlan,
+ carbonTable,
+ agg)
+ Aggregate(updatedGroupExp,
+ updatedAggExp,
+ newChild)
+ } else {
+ agg
}
- carbonTable
- // case for handling aggregation with order by and filter when only projection column exits
- case Sort(sortOrders,
- _,
- Aggregate(groupingExp,
- aggregateExp,
- Filter(filterExp, CarbonSubqueryAlias(_, logicalRelation: LogicalRelation))))
- if logicalRelation.relation.isInstanceOf[CarbonDatasourceHadoopRelation] &&
- logicalRelation.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonRelation.
- metaData.hasAggregateDataMapSchema =>
- val (carbonTable, tableName) = getCarbonTableAndTableName(logicalRelation)
- isValidPlan = extractQueryColumnsFromAggExpression(groupingExp,
- aggregateExp,
+ } else {
+ agg
+ }
+ // case of handling aggregation query with filter
+ case agg@Aggregate(
+ grExp,
+ aggExp,
+ Filter(expression, child@CarbonSubqueryAlias(_, l: LogicalRelation)))
+ if l.relation.isInstanceOf[CarbonDatasourceHadoopRelation] &&
+ l.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonRelation.
+ metaData.hasAggregateDataMapSchema =>
+ val carbonTable = getCarbonTable(l)
+ val list = scala.collection.mutable.HashSet.empty[QueryColumn]
+ val aggregateExpressions = scala.collection.mutable.HashSet.empty[AggregateExpression]
+ var isValidPlan = extractQueryColumnsFromAggExpression(
+ grExp,
+ aggExp,
+ carbonTable,
+ list,
+ aggregateExpressions)
+ // getting the columns from filter expression
+ isValidPlan = !CarbonReflectionUtils.hasPredicateSubquery(expression)
+ if (isValidPlan) {
+ extractColumnFromExpression(expression, list, carbonTable, true)
+ }
+ if(isValidPlan) {
+ val (aggDataMapSchema, childPlan) = getChildDataMapForTransformation(list,
+ aggregateExpressions,
carbonTable,
- tableName,
- list)
- if(isValidPlan) {
- isValidPlan = !CarbonReflectionUtils.hasPredicateSubquery(filterExp)
- }
- if(isValidPlan) {
- list ++ extractQueryColumnForOrderBy(sortOrders = sortOrders,
- carbonTable = carbonTable,
- tableName = tableName)
- isValidPlan = extractQueryColumnFromFilterExp(filterExp, list, carbonTable, tableName)
- }
- carbonTable
- case _ =>
- isValidPlan = false
- null
- }
- if (isValidPlan && null != carbonTable) {
- isValidPlan = isSpecificSegmentPresent(carbonTable)
- }
- // if plan is valid then update the plan with child attributes
- if (isValidPlan) {
- // getting all the projection columns
- val listProjectionColumn = list
- .filter(queryColumn => queryColumn.getAggFunction.isEmpty && !queryColumn.isFilterColumn)
- .toList
- // getting all the filter columns
- val listFilterColumn = list
- .filter(queryColumn => queryColumn.getAggFunction.isEmpty && queryColumn.isFilterColumn)
- .toList
- // getting all the aggregation columns
- val listAggregationColumn = list.filter(queryColumn => !queryColumn.getAggFunction.isEmpty)
- .toList
- // create a query plan object which will be used to select the list of pre aggregate tables
- // matches with this plan
- val queryPlan = new QueryPlan(listProjectionColumn.asJava,
- listAggregationColumn.asJava,
- listFilterColumn.asJava)
- // create aggregate table selector object
- val aggregateTableSelector = new AggregateTableSelector(queryPlan, carbonTable)
- // select the list of valid child tables
- val selectedDataMapSchemas = aggregateTableSelector.selectPreAggDataMapSchema()
- // if it does not match with any pre aggregate table return the same plan
- if (!selectedDataMapSchemas.isEmpty) {
- // filter the selected child schema based on size to select the pre-aggregate tables
- // that are nonEmpty
- val catalog = CarbonEnv.getInstance(sparkSession).carbonMetastore
- val relationBuffer = selectedDataMapSchemas.asScala.map { selectedDataMapSchema =>
- val identifier = TableIdentifier(
- selectedDataMapSchema.getRelationIdentifier.getTableName,
- Some(selectedDataMapSchema.getRelationIdentifier.getDatabaseName))
- val carbonRelation =
- catalog.lookupRelation(identifier)(sparkSession).asInstanceOf[CarbonRelation]
- val relation = sparkSession.sessionState.catalog.lookupRelation(identifier)
- (selectedDataMapSchema, carbonRelation, relation)
- }.filter(_._2.sizeInBytes != 0L)
- if (relationBuffer.isEmpty) {
- // If the size of relation Buffer is 0 then it means that none of the pre-aggregate
- // tables have date yet.
- // In this case we would return the original plan so that the query hits the parent
- // table.
- plan
+ agg)
+ if(null != aggDataMapSchema && null!= childPlan) {
+ val attributes = childPlan.output.asInstanceOf[Seq[AttributeReference]]
+ val (updatedGroupExp, updatedAggExp, newChild, updatedFilterExpression) =
+ getUpdatedExpressions(grExp,
+ aggExp,
+ child,
+ Some(expression),
+ aggDataMapSchema,
+ attributes,
+ childPlan,
+ carbonTable,
+ agg)
+ Aggregate(updatedGroupExp,
+ updatedAggExp,
+ Filter(updatedFilterExpression.get,
+ newChild))
} else {
- // If the relationBuffer is nonEmpty then find the table with the minimum size.
- val (aggDataMapSchema, _, relation) = relationBuffer.minBy(_._2.sizeInBytes)
- val newRelation = new FindDataSourceTable(sparkSession).apply(relation)
- // transform the query plan based on selected child schema
- transformPreAggQueryPlan(plan, aggDataMapSchema, newRelation)
+ agg
}
} else {
- plan
+ agg
}
+ }
+ updatedPlan
+ }
+
+ /**
+ * Below method will be used to validate query plan and get the proper aggregation data map schema
+ * and child relation plan object if plan is valid for transformation
+ * @param queryColumns list of query columns from projection and filter
+ * @param aggregateExpressions list of aggregate expression (aggregate function)
+ * @param carbonTable parent carbon table
+ * @param parentLogicalPlan parent logical relation
+ * @return if plan is valid then aggregation data map schema and its relation plan
+ */
+ def getChildDataMapForTransformation(queryColumns: scala.collection.mutable.HashSet[QueryColumn],
+ aggregateExpressions: scala.collection.mutable.HashSet[AggregateExpression],
+ carbonTable: CarbonTable,
+ parentLogicalPlan: LogicalPlan): (AggregationDataMapSchema, LogicalPlan) = {
+ // getting all the projection columns
+ val listProjectionColumn = queryColumns
+ .filter(queryColumn => !queryColumn.isFilterColumn)
+ .toList
+ // getting all the filter columns
+ val listFilterColumn = queryColumns
+ .filter(queryColumn => queryColumn.isFilterColumn)
+ .toList
+ val isProjectionColumnPresent = (listProjectionColumn.size + listFilterColumn.size) > 0
+ // create a query plan object which will be used to select the list of pre aggregate tables
+ // matches with this plan
+ val queryPlan = new AggregateQueryPlan(listProjectionColumn.asJava, listFilterColumn.asJava)
+ // create aggregate table selector object
+ val aggregateTableSelector = new AggregateTableSelector(queryPlan, carbonTable)
+ // select the list of valid child tables
+ val selectedDataMapSchemas = aggregateTableSelector.selectPreAggDataMapSchema()
+ // query has only aggregate expression then selected data map will be empty
+ // the validate all the child data map otherwise validate selected data map
+ var selectedAggMaps = if (isProjectionColumnPresent) {
+ selectedDataMapSchemas
+ } else {
+ carbonTable.getTableInfo.getDataMapSchemaList
+ }
+ // if it does not match with any pre aggregate table return the same plan
+ if (!selectedAggMaps.isEmpty) {
+ // filter the selected child schema based on size to select the pre-aggregate tables
+ // that are nonEmpty
+ val catalog = CarbonEnv.getInstance(sparkSession).carbonMetastore
+ val relationBuffer = selectedAggMaps.asScala.map { selectedDataMapSchema =>
+ val identifier = TableIdentifier(
+ selectedDataMapSchema.getRelationIdentifier.getTableName,
+ Some(selectedDataMapSchema.getRelationIdentifier.getDatabaseName))
+ val carbonRelation =
+ catalog.lookupRelation(identifier)(sparkSession).asInstanceOf[CarbonRelation]
+ val relation = sparkSession.sessionState.catalog.lookupRelation(identifier)
+ (selectedDataMapSchema, carbonRelation, relation)
+ }.filter(_._2.sizeInBytes != 0L).sortBy(_._2.sizeInBytes)
+ if (relationBuffer.isEmpty) {
+ // If the size of relation Buffer is 0 then it means that none of the pre-aggregate
+ // tables have data yet.
+ // In this case we would return the original plan so that the query hits the parent
+ // table.
+ (null, null)
} else {
- plan
+ // if query does not have any aggregate function no need to validate the same
+ val tuple = if (aggregateExpressions.nonEmpty && !selectedAggMaps.isEmpty) {
+ relationBuffer.collectFirst {
+ case a@(datamapSchema, _, _)
+ if validateAggregateExpression(datamapSchema,
+ carbonTable,
+ parentLogicalPlan,
+ aggregateExpressions.toSeq) =>
+ a
+ }
+ } else {
+ Some(relationBuffer.head)
+ }
+ tuple match {
+ case Some((dataMapSchema, _, logicalPlan)) => (dataMapSchema
+ .asInstanceOf[AggregationDataMapSchema], new FindDataSourceTable(sparkSession)
+ .apply(logicalPlan))
+ case None => (null, null)
+ }
+ // If the relationBuffer is nonEmpty then find the table with the minimum size.
+ }
+ } else {
+ (null, null)
+ }
+ }
+
+ /**
+ * Below method will be used to validate aggregate expression with the data map
+ * and will return the selected valid data maps
+ * @param selectedDataMap list of data maps
+ * @param carbonTable parent carbon table
+ * @param parentLogicalPlan parent logical plan
+ * @param queryAggExpLogicalPlans query agg expression logical plan
+ * @return valid data map
+ */
+ def validateAggregateExpression(selectedDataMap: DataMapSchema,
+ carbonTable: CarbonTable,
+ parentLogicalPlan: LogicalPlan,
+ queryAggExpLogicalPlans: Seq[AggregateExpression]): Boolean = {
+ val mappingModel = getExpressionToColumnMapping(selectedDataMap,
+ carbonTable,
+ parentLogicalPlan)
+ queryAggExpLogicalPlans.forall{p =>
+ mappingModel.exists{m =>
+ PreAggregateUtil.normalizeExprId(p, parentLogicalPlan.allAttributes) == m.expression}
+ }
+ }
+
+ /**
+ * Below method will be used to to get the logical plan for each aggregate expression in
+ * child data map and its column schema mapping if mapping is already present
+ * then it will use the same otherwise it will generate and stored in aggregation data map
+ * @param selectedDataMap child data map
+ * @param carbonTable parent table
+ * @param parentLogicalPlan logical relation of actual plan
+ * @return map of logical plan for each aggregate expression in child query and its column mapping
+ */
+ def getExpressionToColumnMapping(selectedDataMap: DataMapSchema,
+ carbonTable: CarbonTable,
+ parentLogicalPlan: LogicalPlan): mutable.Set[AggExpToColumnMappingModel] = {
+ val aggDataMapSchema = selectedDataMap.asInstanceOf[AggregationDataMapSchema]
+ if(null == aggDataMapSchema.getAggExpToColumnMapping) {
+ // add preAGG UDF to avoid all the PreAggregate rule
+ val childDataMapQueryString = parser.addPreAggFunction(
+ aggDataMapSchema.getProperties.get("CHILD_SELECT QUERY"))
+ // get the logical plan
+ val aggPlan = sparkSession.sql(childDataMapQueryString).logicalPlan
+ // getting all aggregate expression from query
+ val dataMapAggExp = getAggregateExpFromChildDataMap(aggPlan)
+ // in case of average child table will have two columns which will be stored in sequence
+ // so for average expression we need to get two columns for mapping
+ var counter = 0
+ // sorting the columns based on schema ordinal so search will give proper result
+ val sortedColumnList = aggDataMapSchema.getChildSchema.getListOfColumns.asScala
+ .sortBy(_.getSchemaOrdinal)
+ val expressionToColumnMapping = mutable.LinkedHashSet.empty[AggExpToColumnMappingModel]
+ dataMapAggExp.foreach { aggExp =>
+ val updatedExp = PreAggregateUtil.normalizeExprId(aggExp, aggPlan.allAttributes)
+ val model = AggExpToColumnMappingModel(updatedExp, None)
+ if (!expressionToColumnMapping.contains(model)) {
+ // check if aggregate expression is of type avg
+ // get the columns
+ val columnSchema = aggDataMapSchema
+ .getAggColumnBasedOnIndex(counter, sortedColumnList.asJava)
+ // increment the counter so when for next expression above code will be
+ // executed it will search from that schema ordinal
+ counter = columnSchema.getSchemaOrdinal + 1
+ model.columnSchema = Some(columnSchema)
+ expressionToColumnMapping += model
+ }
}
+ aggDataMapSchema.setAggExpToColumnMapping(expressionToColumnMapping.asJava)
+ // return the mapping
+ expressionToColumnMapping
+ } else {
+ aggDataMapSchema.getAggExpToColumnMapping
+ .asInstanceOf[java.util.Set[AggExpToColumnMappingModel]].asScala
+ .asInstanceOf[mutable.LinkedHashSet[AggExpToColumnMappingModel]]
}
}
/**
+ * Below method will be used to get aggregate expression
+ * @param logicalPlan logical plan
+ * @return list of aggregate expression
+ */
+ def getAggregateExpFromChildDataMap(logicalPlan: LogicalPlan): Seq[AggregateExpression] = {
+ val list = scala.collection.mutable.ListBuffer.empty[AggregateExpression]
+ logicalPlan match {
+ case _@Aggregate(_, aggExp, _) =>
+ aggExp map {
+ case Alias(attr: AggregateExpression, _) =>
+ list ++= PreAggregateUtil.validateAggregateFunctionAndGetFields(attr)
+ case _ =>
+ }
+ }
+ list
+ }
+
+ /**
* Below method will be used to check whether specific segment is set for maintable
* if it is present then no need to transform the plan and query will be executed on
* maintable
- * @param carbonTable
- * parent table
+ * @param carbonTable parent table
* @return is specific segment is present in session params
*/
def isSpecificSegmentPresent(carbonTable: CarbonTable) : Boolean = {
@@ -353,26 +554,21 @@ case class CarbonPreAggregateQueryRules(sparkSession: SparkSession) extends Rule
/**
* Below method will be used to extract the query columns from
* filter expression
- * @param filterExp
- * filter expression
- * @param set
- * query column list
- * @param carbonTable
- * parent table
- * @param tableName
- * table name
+ * @param expression filter expression
+ * @param queryColumns query column set
+ * @param carbonTable parent table
* @return isvalid filter expression for aggregate
*/
- def extractQueryColumnFromFilterExp(filterExp: Expression,
- set: scala.collection.mutable.HashSet[QueryColumn],
- carbonTable: CarbonTable, tableName: String): Boolean = {
+ def extractColumnFromExpression(expression: Expression,
+ queryColumns: scala.collection.mutable.HashSet[QueryColumn],
+ carbonTable: CarbonTable,
+ isFilterColumn: Boolean = false) {
// map to maintain attribute reference present in the filter to timeseries function
// if applied this is added to avoid duplicate column
val mapOfColumnSeriesFun = scala.collection.mutable.HashMap.empty[AttributeReference, String]
- val isValidPlan = true
- filterExp.transform {
+ expression.transform {
case attr: AttributeReference =>
- if (!mapOfColumnSeriesFun.get(attr).isDefined) {
+ if (mapOfColumnSeriesFun.get(attr).isEmpty) {
mapOfColumnSeriesFun.put(attr, null)
}
attr
@@ -381,13 +577,13 @@ case class CarbonPreAggregateQueryRules(sparkSession: SparkSession) extends Rule
if (udf.asInstanceOf[ScalaUDF].function.getClass.getName.equalsIgnoreCase(
"org.apache.spark.sql.execution.command.timeseries.TimeseriesFunction") &&
CarbonUtil.hasTimeSeriesDataMap(carbonTable)) {
- mapOfColumnSeriesFun.put(udf.children(0).asInstanceOf[AttributeReference],
- udf.children(1).asInstanceOf[Literal].value.toString)
+ mapOfColumnSeriesFun.put(udf.children.head.asInstanceOf[AttributeReference],
+ udf.children.last.asInstanceOf[Literal].value.toString)
} else {
// for any other scala udf
udf.transform {
case attr: AttributeReference =>
- if (!mapOfColumnSeriesFun.get(attr).isDefined) {
+ if (mapOfColumnSeriesFun.get(attr).isEmpty) {
mapOfColumnSeriesFun.put(attr, null)
}
attr
@@ -396,90 +592,37 @@ case class CarbonPreAggregateQueryRules(sparkSession: SparkSession) extends Rule
udf
}
mapOfColumnSeriesFun.foreach { f =>
- if (f._2 == null) {
- set +=
- getQueryColumn(f._1.name, carbonTable, tableName, isFilterColumn = true)
- } else {
- set += getQueryColumn(f._1.name,
- carbonTable,
- carbonTable.getTableName,
- isFilterColumn = true,
- timeseriesFunction = f._2)
- }
- }
- isValidPlan
- }
- /**
- * Below method will be used to extract columns from order by expression
- * @param projectList
- * project list from plan
- * @param sortOrders
- * sort order in plan
- * @param carbonTable
- * carbon table
- * @param tableName
- * table name
- * @return query columns from expression
- */
- def extractQueryColumnForOrderBy(projectList: Option[Seq[NamedExpression]] = None,
- sortOrders: Seq[SortOrder],
- carbonTable: CarbonTable,
- tableName: String): Seq[QueryColumn] = {
- val list = scala.collection.mutable.ListBuffer.empty[QueryColumn]
- if(projectList.isDefined) {
- projectList.get.map {
- proList =>
- proList.transform {
- case attr: AttributeReference =>
- val queryColumn = getQueryColumn(attr.name, carbonTable, tableName)
- if (null != queryColumn) {
- list += queryColumn
- }
- attr
- }
+ if (f._2 == null) {
+ queryColumns +=
+ getQueryColumn(f._1.name, carbonTable, isFilterColumn)
+ } else {
+ queryColumns += getQueryColumn(f._1.name,
+ carbonTable,
+ isFilterColumn,
+ timeseriesFunction = f._2)
}
}
- sortOrders.foreach { sortOrder =>
- sortOrder.child match {
- case attr: AttributeReference =>
- val queryColumn = getQueryColumn(attr.name, carbonTable, tableName)
- if (null != queryColumn) {
- list += queryColumn
- }
- }
- }
- list
}
/**
* Below method will be used to get the child attribute reference
* based on parent name
*
- * @param dataMapSchema
- * child schema
- * @param attributeReference
- * parent attribute reference
- * @param attributes
- * child logical relation
- * @param aggFunction
- * aggregation function applied on child
- * @param canBeNull
- * this is added for strict validation in which case child attribute can be
+ * @param dataMapSchema child schema
+ * @param attributeReference parent attribute reference
+ * @param attributes child logical relation
+ * @param canBeNull this is added for strict validation in which case child attribute can be
* null and when it cannot be null
* @return child attribute reference
*/
def getChildAttributeReference(dataMapSchema: DataMapSchema,
attributeReference: AttributeReference,
attributes: Seq[AttributeReference],
- aggFunction: String = "",
canBeNull: Boolean = false,
timeseriesFunction: String = ""): AttributeReference = {
- val aggregationDataMapSchema = dataMapSchema.asInstanceOf[AggregationDataMapSchema];
- val columnSchema = if (aggFunction.isEmpty && timeseriesFunction.isEmpty) {
+ val aggregationDataMapSchema = dataMapSchema.asInstanceOf[AggregationDataMapSchema]
+ val columnSchema = if (timeseriesFunction.isEmpty) {
aggregationDataMapSchema.getChildColByParentColName(attributeReference.name.toLowerCase)
- } else if (!aggFunction.isEmpty) {
- aggregationDataMapSchema.getAggChildColByParent(attributeReference.name.toLowerCase,
- aggFunction)
} else {
aggregationDataMapSchema
.getTimeseriesChildColByParent(attributeReference.name.toLowerCase,
@@ -499,254 +642,6 @@ case class CarbonPreAggregateQueryRules(sparkSession: SparkSession) extends Rule
}
/**
- * Below method will be used to transform the main table plan to child table plan
- * rules for transformming is as below.
- * 1. Grouping expression rules
- * 1.1 Change the parent attribute reference for of group expression
- * to child attribute reference
- *
- * 2. Aggregate expression rules
- * 2.1 Change the parent attribute reference for of group expression to
- * child attribute reference
- * 2.2 Change the count AggregateExpression to Sum as count
- * is already calculated so in case of aggregate table
- * we need to apply sum to get the count
- * 2.2 In case of average aggregate function select 2 columns from aggregate table with
- * aggregation sum and count. Then add divide(sum(column with sum), sum(column with count)).
- * Note: During aggregate table creation for average table will be created with two columns
- * one for sum(column) and count(column) to support rollup
- * 3. Filter Expression rules.
- * 3.1 Updated filter expression attributes with child table attributes
- * 4. Update the Parent Logical relation with child Logical relation
- * 5. Order by plan rules.
- * 5.1 Update project list based on updated aggregate expression
- * 5.2 Update sort order attributes based on pre aggregate table
- * 6. timeseries function
- * 6.1 validate parent table has timeseries datamap
- * 6.2 timeseries function is valid function or not
- *
- * @param logicalPlan
- * parent logical plan
- * @param aggDataMapSchema
- * select data map schema
- * @param childPlan
- * child carbon table relation
- * @return transformed plan
- */
- def transformPreAggQueryPlan(logicalPlan: LogicalPlan,
- aggDataMapSchema: DataMapSchema,
- childPlan: LogicalPlan): LogicalPlan = {
- val attributes = childPlan.output.asInstanceOf[Seq[AttributeReference]]
- logicalPlan.transform {
- // case for aggregation query
- case Aggregate(grExp, aggExp, child@CarbonSubqueryAlias(_, l: LogicalRelation))
- if l.relation.isInstanceOf[CarbonDatasourceHadoopRelation] &&
- l.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonRelation.
- metaData.hasAggregateDataMapSchema =>
- val (updatedGroupExp, updatedAggExp, newChild, None) =
- getUpdatedExpressions(grExp,
- aggExp,
- child,
- None,
- aggDataMapSchema,
- attributes,
- childPlan)
- Aggregate(updatedGroupExp,
- updatedAggExp,
- newChild)
- // case of handling aggregation query with filter
- case Aggregate(grExp,
- aggExp,
- Filter(expression, child@CarbonSubqueryAlias(_, l: LogicalRelation)))
- if l.relation.isInstanceOf[CarbonDatasourceHadoopRelation] &&
- l.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonRelation.
- metaData.hasAggregateDataMapSchema =>
- val (updatedGroupExp, updatedAggExp, newChild, updatedFilterExpression) =
- getUpdatedExpressions(grExp,
- aggExp,
- child,
- Some(expression),
- aggDataMapSchema,
- attributes,
- childPlan)
- Aggregate(updatedGroupExp,
- updatedAggExp,
- Filter(updatedFilterExpression.get,
- newChild))
- // case for aggregation query
- case Aggregate(grExp, aggExp, l: LogicalRelation)
- if l.relation.isInstanceOf[CarbonDatasourceHadoopRelation] &&
- l.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonRelation.
- metaData.hasAggregateDataMapSchema =>
- val (updatedGroupExp, updatedAggExp, newChild, None) =
- getUpdatedExpressions(grExp,
- aggExp,
- l,
- None,
- aggDataMapSchema,
- attributes,
- childPlan)
- Aggregate(updatedGroupExp,
- updatedAggExp,
- newChild)
- // case for aggregation query with order by
- case Project(_,
- Sort(sortOrders,
- global,
- Aggregate(groupingExp,
- aggregateExp,
- subQuery@CarbonSubqueryAlias(_, l: LogicalRelation))))
- if l.relation.isInstanceOf[CarbonDatasourceHadoopRelation] &&
- l.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonRelation.
- metaData.hasAggregateDataMapSchema =>
- val (updatedGroupExp, updatedAggExp, newChild, None) =
- getUpdatedExpressions(groupingExp,
- aggregateExp,
- subQuery,
- None,
- aggDataMapSchema,
- attributes,
- childPlan)
- val (updatedProjectList, updatedSortOrder) = transformPlanForOrderBy(updatedAggExp,
- sortOrders,
- aggDataMapSchema,
- attributes)
- Project(updatedProjectList,
- Sort(updatedSortOrder, global, Aggregate(updatedGroupExp, updatedAggExp, newChild)))
- // case for handling aggregation query with filter and order by
- case Project(_,
- Sort(sortOrders,
- global,
- Aggregate(groupingExp,
- aggregateExp,
- Filter(expression, subQuery@CarbonSubqueryAlias(_, l: LogicalRelation)))))
- if l.relation.isInstanceOf[CarbonDatasourceHadoopRelation] &&
- l.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonRelation.
- metaData.hasAggregateDataMapSchema =>
- val (updatedGroupExp, updatedAggExp, newChild, updatedFilterExpression) =
- getUpdatedExpressions(groupingExp,
- aggregateExp,
- subQuery,
- Some(expression),
- aggDataMapSchema,
- attributes,
- childPlan)
- val (updatedProjectList, updatedSortOrder) = transformPlanForOrderBy(updatedAggExp,
- sortOrders,
- aggDataMapSchema,
- attributes)
- Project(updatedProjectList,
- Sort(updatedSortOrder, global, Aggregate(updatedGroupExp, updatedAggExp,
- Filter(updatedFilterExpression.get, newChild))))
- // case for handling aggregation with order by when only projection column exits
- case Sort(sortOrders,
- global,
- Aggregate(
- groupingExp,
- aggregateExp,
- subQuery@CarbonSubqueryAlias(_, logicalRelation: LogicalRelation)))
- if logicalRelation.relation.isInstanceOf[CarbonDatasourceHadoopRelation] &&
- logicalRelation.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonRelation.
- metaData.hasAggregateDataMapSchema =>
- val (updatedGroupExp, updatedAggExp, newChild, None) =
- getUpdatedExpressions(groupingExp,
- aggregateExp,
- subQuery,
- None,
- aggDataMapSchema,
- attributes,
- childPlan)
- val (_, updatedSortOrder) = transformPlanForOrderBy(updatedAggExp,
- sortOrders,
- aggDataMapSchema,
- attributes)
- Sort(updatedSortOrder, global, Aggregate(updatedGroupExp, updatedAggExp, newChild))
- // case for handling aggregation with order by and filter when only projection column exits
- case Sort(sortOrders,
- global,
- Aggregate(groupingExp,
- aggregateExp,
- Filter(expression, subQuery@CarbonSubqueryAlias(_, l: LogicalRelation))))
- if l.relation.isInstanceOf[CarbonDatasourceHadoopRelation] &&
- l.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonRelation.
- metaData.hasAggregateDataMapSchema =>
- val (updatedGroupExp, updatedAggExp, newChild, updatedFilterExpression) =
- getUpdatedExpressions(groupingExp,
- aggregateExp,
- subQuery,
- Some(expression),
- aggDataMapSchema,
- attributes,
- childPlan)
- val (_, updatedSortOrder) = transformPlanForOrderBy(updatedAggExp,
- sortOrders,
- aggDataMapSchema,
- attributes)
- Sort(updatedSortOrder, global, Aggregate(updatedGroupExp, updatedAggExp, newChild))
- }
- }
-
- /**
- * Below method will be used to updated the maintable plan for order by query
- * In case of order by we need to update project list and sort order attributes.
- *
- * @param aggregateExp
- * child table aggregate expression
- * @param sortOrders
- * sort order expression in maintable plan
- * @param aggDataMapSchema
- * child data map schema
- * @param attributes
- * child attributes
- * @return updated project list and updated sort order
- */
- def transformPlanForOrderBy(aggregateExp: Seq[NamedExpression],
- sortOrders: Seq[SortOrder], aggDataMapSchema: DataMapSchema,
- attributes: Seq[AttributeReference]): (Seq[NamedExpression], Seq[SortOrder]) = {
- val updatedProjectList = new ArrayBuffer[NamedExpression]()
- // getting the updated project list from aggregate expression
- aggregateExp.foreach{f => f.transform {
- // for projection column
- case alias@Alias(attr: AttributeReference, name) =>
- updatedProjectList += AttributeReference(name, attr.dataType, attr.nullable)(alias.exprId,
- alias.qualifier,
- alias.isGenerated)
- alias
- // for aggregaton column
- case alias@Alias(attr: AggregateExpression, name) =>
- updatedProjectList += AttributeReference(name, attr.dataType, attr.nullable)(alias.exprId,
- alias.qualifier,
- alias.isGenerated)
- alias
- case alias@Alias(exp: Expression, name) =>
- updatedProjectList += AttributeReference(name, exp.dataType, exp.nullable)(alias.exprId,
- alias.qualifier,
- alias.isGenerated)
- alias
- }
- }
- // getting the updated sort order
- val updatedSortOrders = sortOrders.map { order =>
- order.child match {
- case attr: AttributeReference =>
- val childAttribute = getChildAttributeReference(aggDataMapSchema,
- attr,
- attributes,
- canBeNull = true)
- // child attribute can be null only in case of alias in query
- // so in that case we need to update the sortorder based on new alias
- if (null != childAttribute) {
- val childExpression = getUpdatedSortOrderExpression(childAttribute, aggregateExp)
- SortOrder(childExpression, order.direction)
- } else {
- val childExpression = getUpdatedSortOrderExpression(attr, aggregateExp)
- SortOrder(childExpression, order.direction)
- }
- }
- }
- (updatedProjectList, updatedSortOrders)
- }
- /**
* Below method will be used to get the updated expression for pre aggregated table.
* It will replace the attribute of actual plan with child table attributes.
* Updation will be done for below expression.
@@ -755,37 +650,41 @@ case class CarbonPreAggregateQueryRules(sparkSession: SparkSession) extends Rule
* 3. child logical plan
* 4. filter expression if present
*
- * @param groupingExpressions
- * actual plan grouping expression
- * @param aggregateExpressions
- * actual plan aggregate expression
- * @param child
- * child logical plan
- * @param filterExpression
- * filter expression
- * @param aggDataMapSchema
- * pre aggregate table schema
- * @param attributes
- * pre aggregate table logical relation
- * @param aggPlan
- * aggregate logical plan
+ * @param groupingExpressions actual plan grouping expression
+ * @param aggregateExpressions actual plan aggregate expression
+ * @param child child logical plan
+ * @param filterExpression filter expression
+ * @param aggDataMapSchema pre aggregate table schema
+ * @param attributes pre aggregate table logical relation
+ * @param aggPlan aggregate logical plan
* @return tuple of(updated grouping expression,
- * updated aggregate expression,
- * updated child logical plan,
- * updated filter expression if present in actual plan)
+ * updated aggregate expression,
+ * updated child logical plan,
+ * updated filter expression if present in actual plan)
*/
def getUpdatedExpressions(groupingExpressions: Seq[Expression],
aggregateExpressions: Seq[NamedExpression],
child: LogicalPlan, filterExpression: Option[Expression] = None,
- aggDataMapSchema: DataMapSchema,
+ aggDataMapSchema: AggregationDataMapSchema,
attributes: Seq[AttributeReference],
- aggPlan: LogicalPlan): (Seq[Expression], Seq[NamedExpression], LogicalPlan,
- Option[Expression]) = {
+ aggPlan: LogicalPlan,
+ parentTable: CarbonTable,
+ parentLogicalPlan: LogicalPlan): (Seq[Expression], Seq[NamedExpression], LogicalPlan,
+ Option[Expression]) = {
+ val aggExpColumnMapping = if (null != aggDataMapSchema.getAggExpToColumnMapping) {
+ Some(aggDataMapSchema.getAggExpToColumnMapping
+ .asInstanceOf[java.util.Set[AggExpToColumnMappingModel]].asScala
+ .asInstanceOf[mutable.LinkedHashSet[AggExpToColumnMappingModel]])
+ } else {
+ None
+ }
+
// transforming the group by expression attributes with child attributes
val updatedGroupExp = groupingExpressions.map { exp =>
exp.transform {
case attr: AttributeReference =>
- getChildAttributeReference(aggDataMapSchema, attr, attributes)
+ val childAttr = getChildAttributeReference(aggDataMapSchema, attr, attributes)
+ childAttr
}
}
// below code is for updating the aggregate expression.
@@ -806,59 +705,97 @@ case class CarbonPreAggregateQueryRules(sparkSession: SparkSession) extends Rule
val updatedAggExp = aggregateExpressions.map {
// case for attribute reference
case attr: AttributeReference =>
- val childAttributeReference = getChildAttributeReference(aggDataMapSchema,
+ val childAttr = getChildAttributeReference(aggDataMapSchema,
attr,
attributes)
+ val newExpressionId = NamedExpression.newExprId
+ val childTableAttr = AttributeReference(attr.name,
+ childAttr.dataType,
+ childAttr.nullable,
+ childAttr.metadata)(newExpressionId, childAttr.qualifier, attr.isGenerated)
+ updatedExpression.put(attr, childTableAttr)
// returning the alias to show proper column name in output
- Alias(childAttributeReference,
- attr.name)(NamedExpression.newExprId,
- childAttributeReference.qualifier).asInstanceOf[NamedExpression]
+ Alias(childAttr,
+ attr.name)(newExpressionId,
+ childAttr.qualifier).asInstanceOf[NamedExpression]
// case for alias
- case Alias(attr: AttributeReference, name) =>
- val childAttributeReference = getChildAttributeReference(aggDataMapSchema,
+ case alias@Alias(attr: AttributeReference, name) =>
+ val childAttr = getChildAttributeReference(aggDataMapSchema,
attr,
attributes)
+ val newExpressionId = NamedExpression.newExprId
+ val parentTableAttr = AttributeReference(name,
+ alias.dataType,
+ alias.nullable) (alias.exprId, alias.qualifier, alias.isGenerated)
+ val childTableAttr = AttributeReference(name,
+ alias.dataType,
+ alias.nullable) (newExpressionId, alias.qualifier, alias.isGenerated)
+ updatedExpression.put(parentTableAttr, childTableAttr)
// returning alias with child attribute reference
- Alias(childAttributeReference,
- name)(NamedExpression.newExprId,
- childAttributeReference.qualifier).asInstanceOf[NamedExpression]
+ Alias(childAttr,
+ name)(newExpressionId,
+ childAttr.qualifier).asInstanceOf[NamedExpression]
// for aggregate function case
case alias@Alias(attr: AggregateExpression, name) =>
// get the updated aggregate aggregate function
- val aggExp = getUpdatedAggregateExpressionForChild(attr,
- aggDataMapSchema,
- attributes)
+ val aggExp = if (aggExpColumnMapping.isDefined) {
+ getUpdatedAggregateExpressionForChild(attr,
+ aggDataMapSchema,
+ attributes,
+ parentTable,
+ parentLogicalPlan,
+ aggExpColumnMapping.get)
+ } else {
+ attr
+ }
+ val newExpressionId = NamedExpression.newExprId
+ val parentTableAttr = AttributeReference(name,
+ alias.dataType,
+ alias.nullable) (alias.exprId, alias.qualifier, alias.isGenerated)
+ val childTableAttr = AttributeReference(name,
+ alias.dataType,
+ alias.nullable) (newExpressionId, alias.qualifier, alias.isGenerated)
+ updatedExpression.put(parentTableAttr, childTableAttr)
// returning alias with child attribute reference
Alias(aggExp,
- name)(NamedExpression.newExprId,
+ name)(newExpressionId,
alias.qualifier).asInstanceOf[NamedExpression]
case alias@Alias(expression: Expression, name) =>
val updatedExp =
if (expression.isInstanceOf[ScalaUDF] &&
expression.asInstanceOf[ScalaUDF].function.getClass.getName.equalsIgnoreCase(
"org.apache.spark.sql.execution.command.timeseries.TimeseriesFunction")) {
- expression.asInstanceOf[ScalaUDF].transform {
- case attr: AttributeReference =>
- val childAttributeReference = getChildAttributeReference(aggDataMapSchema,
- attr,
- attributes,
- timeseriesFunction =
- expression.asInstanceOf[ScalaUDF].children(1).asInstanceOf[Literal].value.toString)
- childAttributeReference
- }
- } else {
- expression.transform{
- case attr: AttributeReference =>
- val childAttributeReference = getChildAttributeReference(aggDataMapSchema,
- attr,
- attributes)
- childAttributeReference
+ expression.asInstanceOf[ScalaUDF].transform {
+ case attr: AttributeReference =>
+ val childAttributeReference = getChildAttributeReference(aggDataMapSchema,
+ attr,
+ attributes,
+ timeseriesFunction =
+ expression.asInstanceOf[ScalaUDF].children(1).asInstanceOf[Literal].value
+ .toString)
+ childAttributeReference
+ }
+ } else {
+ expression.transform{
+ case attr: AttributeReference =>
+ val childAttributeReference = getChildAttributeReference(aggDataMapSchema,
+ attr,
+ attributes)
+ childAttributeReference
+ }
}
- }
- Alias(updatedExp, name)(NamedExpression.newExprId,
+ val newExpressionId = NamedExpression.newExprId
+ val parentTableAttr = AttributeReference(name,
+ alias.dataType,
+ alias.nullable) (alias.exprId, alias.qualifier, alias.isGenerated)
+ val childTableAttr = AttributeReference(name,
+ alias.dataType,
+ alias.nullable) (newExpressionId, alias.qualifier, alias.isGenerated)
+ updatedExpression.put(parentTableAttr, childTableAttr)
+ Alias(updatedExp, name)(newExpressionId,
alias.qualifier).asInstanceOf[NamedExpression]
}
- // transformaing the logical relation
+ // transforming the logical relation
val newChild = child.transform {
case _: LogicalRelation =>
aggPlan
@@ -866,7 +803,7 @@ case class CarbonPreAggregateQueryRules(sparkSession: SparkSession) extends Rule
aggPlan match {
case s: SubqueryAlias => s.child
case others => others
- }
+ }
}
// updating the filter expression if present
val updatedFilterExpression = if (filterExpression.isDefined) {
@@ -882,39 +819,6 @@ case class CarbonPreAggregateQueryRules(sparkSession: SparkSession) extends Rule
}
/**
- * Below method will be used to get the updated sort order attribute
- * based on pre aggregate table
- * @param sortOrderAttr
- * sort order attributes reference
- * @param aggregateExpressions
- * aggregate expression
- * @return updated sortorder attribute
- */
- def getUpdatedSortOrderExpression(sortOrderAttr: AttributeReference,
- aggregateExpressions: Seq[NamedExpression]): Expression = {
- val updatedExpression = aggregateExpressions collectFirst {
- // in case of alias we need to match with alias name and when alias is not present
- // we need to compare with attribute reference name
- case alias@Alias(attr: AttributeReference, name)
- if attr.name.equalsIgnoreCase(sortOrderAttr.name) ||
- name.equalsIgnoreCase(sortOrderAttr.name) =>
- AttributeReference(name,
- sortOrderAttr.dataType,
- sortOrderAttr.nullable,
- sortOrderAttr.metadata)(alias.exprId, alias.qualifier, alias.isGenerated)
- case alias@Alias(_: Expression, name)
- if name.equalsIgnoreCase(sortOrderAttr.name) =>
- AttributeReference(name,
- sortOrderAttr.dataType,
- sortOrderAttr.nullable,
- sortOrderAttr.metadata)(alias.exprId, alias.qualifier, alias.isGenerated)
- }
- // any case it will match the condition, so no need to check whether updated expression is empty
- // or not
- updatedExpression.get
- }
-
- /**
* Below method will be used to get the aggregate expression based on match
* Aggregate expression updation rules
* 1 Change the count AggregateExpression to Sum as count
@@ -927,166 +831,120 @@ case class CarbonPreAggregateQueryRules(sparkSession: SparkSession) extends Rule
* table will be created with two columns one for sum(column) and count(column)
* to support rollup
*
- * @param aggExp
- * aggregate expression
- * @param dataMapSchema
- * child data map schema
- * @param attributes
- * child logical relation
+ * @param aggExp aggregate expression
+ * @param dataMapSchema child data map schema
+ * @param attributes child logical relation
+ * @param parentTable parent carbon table
+ * @param parentLogicalPlan logical relation
* @return updated expression
*/
def getUpdatedAggregateExpressionForChild(aggExp: AggregateExpression,
- dataMapSchema: DataMapSchema,
- attributes: Seq[AttributeReference]):
+ dataMapSchema: AggregationDataMapSchema,
+ attributes: Seq[AttributeReference],
+ parentTable: CarbonTable,
+ parentLogicalPlan: LogicalPlan,
+ aggExpColumnMapping: mutable.LinkedHashSet[AggExpToColumnMappingModel]):
Expression = {
+ // get the updated aggregate expression, in case of average column
+ // it will be divided in two aggergate expression
+ val updatedAggExp = PreAggregateUtil.validateAggregateFunctionAndGetFields(aggExp)
+ // get the attributes to be updated for child table
+ val attrs = aggExpColumnMapping.collect {
+ case (schemaAggExpModel)
+ if updatedAggExp
+ .exists(p =>
+ schemaAggExpModel.expression ==
+ PreAggregateUtil.normalizeExprId(p, parentLogicalPlan.allAttributes)) =>
+ attributes filter (_.name.equalsIgnoreCase(
+ schemaAggExpModel.columnSchema.get.asInstanceOf[ColumnSchema].getColumnName))
+ }.flatten
+
aggExp.aggregateFunction match {
+ case Sum(MatchCastExpression(_, changeDataType: DataType)) =>
+ AggregateExpression(Sum(Cast(attrs.head, changeDataType)), aggExp.mode, isDistinct = false)
+ case Sum(_) =>
+ AggregateExpression(Sum(attrs.head), aggExp.mode, isDistinct = false)
+ case Max(MatchCastExpression(_, changeDataType: DataType)) =>
+ AggregateExpression(Max(Cast(attrs.head, changeDataType)), aggExp.mode, isDistinct = false)
+ case Max(_) =>
+ AggregateExpression(Max(attrs.head), aggExp.mode, isDistinct = false)
+ case Min(MatchCastExpression(_, changeDataType: DataType)) =>
+ AggregateExpression(Min(Cast(attrs.head, changeDataType)), aggExp.mode, isDistinct = false)
+ case Min(_) =>
+ AggregateExpression(Min(attrs.head), aggExp.mode, isDistinct = false)
// Change the count AggregateExpression to Sum as count
// is already calculated so in case of aggregate table
// we need to apply sum to get the count
- case count@Count(Seq(attr: AttributeReference)) =>
- AggregateExpression(Sum(Cast(getChildAttributeReference(dataMapSchema,
- attr,
- attributes,
- count.prettyName),
- LongType)),
- aggExp.mode,
- isDistinct = false)
- case sum@Sum(attr: AttributeReference) =>
- AggregateExpression(Sum(getChildAttributeReference(dataMapSchema,
- attr,
- attributes,
- sum.prettyName)),
- aggExp.mode,
- isDistinct = false)
- case max@Max(attr: AttributeReference) =>
- AggregateExpression(Max(getChildAttributeReference(dataMapSchema,
- attr,
- attributes,
- max.prettyName)),
- aggExp.mode,
- isDistinct = false)
- case min@Min(attr: AttributeReference) =>
- AggregateExpression(Min(getChildAttributeReference(dataMapSchema,
- attr,
- attributes,
- min.prettyName)),
- aggExp.mode,
- isDistinct = false)
- case sum@Sum(MatchCast(attr: AttributeReference, changeDataType: DataType)) =>
- AggregateExpression(Sum(Cast(getChildAttributeReference(dataMapSchema,
- attr,
- attributes,
- sum.prettyName),
- changeDataType)),
- aggExp.mode,
- isDistinct = false)
- case min@Min(MatchCast(attr: AttributeReference, changeDataType: DataType)) =>
- AggregateExpression(Min(Cast(getChildAttributeReference(dataMapSchema,
- attr,
- attributes,
- min.prettyName),
- changeDataType)),
- aggExp.mode,
- isDistinct = false)
- case max@Max(MatchCast(attr: AttributeReference, changeDataType: DataType)) =>
- AggregateExpression(Max(Cast(getChildAttributeReference(dataMapSchema,
- attr,
- attributes,
- max.prettyName),
- changeDataType)),
- aggExp.mode,
- isDistinct = false)
-
+ case Count(Seq(expression: Expression)) =>
+ AggregateExpression(Sum(Cast(attrs.head, LongType)), aggExp.mode, isDistinct = false)
// In case of average aggregate function select 2 columns from aggregate table
// with aggregation sum and count.
// Then add divide(sum(column with sum), sum(column with count)).
- case Average(attr: AttributeReference) =>
- Divide(AggregateExpression(Sum(getChildAttributeReference(dataMapSchema,
- attr,
- attributes,
- "sum")),
+ case Average(MatchCastExpression(exp: Expression, changeDataType: DataType)) =>
+ Divide(AggregateExpression(Sum(Cast(
+ attrs.head,
+ DoubleType)),
aggExp.mode,
isDistinct = false),
- AggregateExpression(Sum(getChildAttributeReference(dataMapSchema,
- attr,
- attributes,
- "count")),
+ AggregateExpression(Sum(Cast(
+ attrs.last,
+ DoubleType)),
aggExp.mode,
isDistinct = false))
// In case of average aggregate function select 2 columns from aggregate table
// with aggregation sum and count.
// Then add divide(sum(column with sum), sum(column with count)).
- case Average(MatchCast(attr: AttributeReference, changeDataType: DataType)) =>
- Divide(AggregateExpression(Sum(Cast(getChildAttributeReference(dataMapSchema,
- attr,
- attributes,
- "sum"),
- DoubleType)),
+ case Average(exp: Expression) =>
+ Divide(AggregateExpression(Sum(attrs.head),
aggExp.mode,
isDistinct = false),
- AggregateExpression(Sum(Cast(getChildAttributeReference(dataMapSchema,
- attr,
- attributes,
- "count"),
- DoubleType)),
+ AggregateExpression(Sum(attrs.last),
aggExp.mode,
isDistinct = false))
}
}
-
/**
* Method to get the carbon table and table name
- *
- * @param parentLogicalRelation
- * parent table relation
- * @return tuple of carbon table and table name
+ * @param parentLogicalRelation parent table relation
+ * @return tuple of carbon table
*/
- def getCarbonTableAndTableName(parentLogicalRelation: LogicalRelation): (CarbonTable, String) = {
+ def getCarbonTable(parentLogicalRelation: LogicalRelation): CarbonTable = {
val carbonTable = parentLogicalRelation.relation.asInstanceOf[CarbonDatasourceHadoopRelation]
.carbonRelation
.metaData.carbonTable
- val tableName = carbonTable.getAbsoluteTableIdentifier.getCarbonTableIdentifier
- .getTableName
- (carbonTable, tableName)
+ carbonTable
}
/**
* Below method will be used to get the query columns from plan
- *
- * @param groupByExpression
- * group by expression
- * @param aggregateExpressions
- * aggregate expression
- * @param carbonTable
- * parent carbon table
- * @param tableName
- * parent table name
- * @param set
- * list of attributes
+ * @param groupByExpression group by expression
+ * @param aggregateExpressions aggregate expression
+ * @param carbonTable parent carbon table
+ * @param queryColumns list of attributes
* @return plan is valid
*/
def extractQueryColumnsFromAggExpression(groupByExpression: Seq[Expression],
aggregateExpressions: Seq[NamedExpression],
- carbonTable: CarbonTable, tableName: String,
- set: scala.collection.mutable.HashSet[QueryColumn]): Boolean = {
+ carbonTable: CarbonTable,
+ queryColumns: scala.collection.mutable.HashSet[QueryColumn],
+ aggreagteExps: scala.collection.mutable.HashSet[AggregateExpression]): Boolean = {
+ groupByExpression foreach { expression =>
+ extractColumnFromExpression(expression, queryColumns, carbonTable)
+ }
aggregateExpressions.map {
case attr: AttributeReference =>
- set += getQueryColumn(attr.name,
- carbonTable,
- tableName)
+ queryColumns += getQueryColumn(attr.name,
+ carbonTable)
case Alias(attr: AttributeReference, _) =>
- set += getQueryColumn(attr.name,
- carbonTable,
- tableName);
+ queryColumns += getQueryColumn(attr.name,
+ carbonTable);
case Alias(attr: AggregateExpression, _) =>
if (attr.isDistinct) {
return false
}
- val queryColumn = validateAggregateFunctionAndGetFields(carbonTable,
- attr.aggregateFunction,
- tableName)
- if (queryColumn.nonEmpty) {
- set ++= queryColumn
+ val aggExp = PreAggregateUtil.validateAggregateFunctionAndGetFields(attr)
+ if (aggExp.nonEmpty) {
+ aggreagteExps ++= aggExp
} else {
return false
}
@@ -1095,18 +953,16 @@ case class CarbonPreAggregateQueryRules(sparkSession: SparkSession) extends Rule
expression.asInstanceOf[ScalaUDF].function.getClass.getName.equalsIgnoreCase(
"org.apache.spark.sql.execution.command.timeseries.TimeseriesFunction") &&
CarbonUtil.hasTimeSeriesDataMap(carbonTable)) {
- set += getQueryColumn(expression.asInstanceOf[ScalaUDF].children(0)
+ queryColumns += getQueryColumn(expression.asInstanceOf[ScalaUDF].children(0)
.asInstanceOf[AttributeReference].name,
carbonTable,
- tableName,
timeseriesFunction = expression.asInstanceOf[ScalaUDF].children(1).asInstanceOf[Literal]
.value.toString)
} else {
expression.transform {
case attr: AttributeReference =>
- set += getQueryColumn(attr.name,
- carbonTable,
- tableName)
+ queryColumns += getQueryColumn(attr.name,
+ carbonTable)
attr
}
}
@@ -1115,215 +971,112 @@ case class CarbonPreAggregateQueryRules(sparkSession: SparkSession) extends Rule
}
/**
- * Below method will be used to validate aggregate function and get the attribute information
- * which is applied on select query.
- * Currently sum, max, min, count, avg is supported
- * in case of any other aggregate function it will return empty sequence
- * In case of avg it will return two fields one for count
- * and other of sum of that column to support rollup
- *
- * @param carbonTable
- * parent table
- * @param aggFunctions
- * aggregation function
- * @param tableName
- * parent table name
- * @return list of fields
- */
- def validateAggregateFunctionAndGetFields(carbonTable: CarbonTable,
- aggFunctions: AggregateFunction,
- tableName: String
- ): Seq[QueryColumn] = {
- val changedDataType = true
- aggFunctions match {
- case sum@Sum(attr: AttributeReference) =>
- Seq(getQueryColumn(attr.name,
- carbonTable,
- tableName,
- sum.prettyName))
- case sum@Sum(MatchCast(attr: AttributeReference, changeDataType: DataType)) =>
- Seq(getQueryColumn(attr.name,
- carbonTable,
- tableName,
- sum.prettyName,
- changeDataType.typeName,
- changedDataType))
- case count@Count(Seq(attr: AttributeReference)) =>
- Seq(getQueryColumn(attr.name,
- carbonTable,
- tableName,
- count.prettyName))
- case min@Min(attr: AttributeReference) =>
- Seq(getQueryColumn(attr.name,
- carbonTable,
- tableName,
- min.prettyName))
- case min@Min(MatchCast(attr: AttributeReference, changeDataType: DataType)) =>
- Seq(getQueryColumn(attr.name,
- carbonTable,
- tableName,
- min.prettyName,
- changeDataType.typeName,
- changedDataType))
- case max@Max(attr: AttributeReference) =>
- Seq(getQueryColumn(attr.name,
- carbonTable,
- tableName,
- max.prettyName))
- case max@Max(MatchCast(attr: AttributeReference, changeDataType: DataType)) =>
- Seq(getQueryColumn(attr.name,
- carbonTable,
- tableName,
- max.prettyName,
- changeDataType.typeName,
- changedDataType))
- // in case of average need to return two columns
- // sum and count of the column to added during table creation to support rollup
- case Average(attr: AttributeReference) =>
- Seq(getQueryColumn(attr.name,
- carbonTable,
- tableName,
- "sum"
- ), getQueryColumn(attr.name,
- carbonTable,
- tableName,
- "count"
- ))
- // in case of average need to return two columns
- // sum and count of the column to added during table creation to support rollup
- case Average(MatchCast(attr: AttributeReference, changeDataType: DataType)) =>
- Seq(getQueryColumn(attr.name,
- carbonTable,
- tableName,
- "sum",
- changeDataType.typeName,
- changedDataType), getQueryColumn(attr.name,
- carbonTable,
- tableName,
- "count",
- changeDataType.typeName,
- changedDataType))
- case _ =>
- Seq.empty
- }
- }
-
-
-
- /**
* Below method will be used to get the query column object which
* will have details of the column and its property
*
- * @param columnName
- * parent column name
- * @param carbonTable
- * parent carbon table
- * @param tableName
- * parent table name
- * @param aggFunction
- * aggregate function applied
- * @param dataType
- * data type of the column
- * @param isChangedDataType
- * is cast is applied on column
- * @param isFilterColumn
- * is filter is applied on column
+ * @param columnName parent column name
+ * @param carbonTable parent carbon table
+ * @param isFilterColumn is filter is applied on column
* @return query column
*/
def getQueryColumn(columnName: String,
carbonTable: CarbonTable,
- tableName: String,
- aggFunction: String = "",
- dataType: String = "",
- isChangedDataType: Boolean = false,
isFilterColumn: Boolean = false,
timeseriesFunction: String = ""): QueryColumn = {
- val columnSchema = carbonTable.getColumnByName(tableName, columnName.toLowerCase)
+ val columnSchema = carbonTable.getColumnByName(carbonTable.getTableName, columnName.toLowerCase)
if(null == columnSchema) {
null
} else {
- if (isChangedDataType) {
- new QueryColumn(columnSchema.getColumnSchema,
- columnSchema.getDataType.getName,
- aggFunction.toLowerCase,
- isFilterColumn,
- timeseriesFunction.toLowerCase)
- } else {
new QueryColumn(columnSchema.getColumnSchema,
- CarbonScalaUtil.convertSparkToCarbonSchemaDataType(dataType),
- aggFunction.toLowerCase,
isFilterColumn,
timeseriesFunction.toLowerCase)
- }
}
}
}
-object CarbonPreAggregateDataLoadingRules extends Rule[LogicalPlan] {
-
+/**
+ * Data loading rule class to validate and update the data loading query plan
+ * Validation rule:
+ * 1. update the avg aggregate expression with two columns sum and count
+ * 2. Remove duplicate sum and count expression if already there in plan
+ * @param sparkSession spark session
+ */
+case class CarbonPreAggregateDataLoadingRules(sparkSession: SparkSession)
+ extends Rule[LogicalPlan] {
+ lazy val parser = new CarbonSpark2SqlParser
override def apply(plan: LogicalPlan): LogicalPlan = {
- val validExpressionsMap = scala.collection.mutable.LinkedHashMap.empty[String, NamedExpression]
+ val validExpressionsMap = scala.collection.mutable.HashSet.empty[AggExpToColumnMappingModel]
+ val namedExpressionList = scala.collection.mutable.LinkedHashSet.empty[NamedExpression]
plan transform {
- case aggregate@Aggregate(_, aExp, _) if validateAggregateExpressions(aExp) =>
+ case aggregate@Aggregate(_,
+ aExp,
+ CarbonSubqueryAlias(_, logicalRelation: LogicalRelation))
+ if validateAggregateExpressions(aExp) &&
+ logicalRelation.relation.isInstanceOf[CarbonDatasourceHadoopRelation] =>
aExp.foreach {
- case alias: Alias =>
- validExpressionsMap ++= validateAggregateFunctionAndGetAlias(alias)
- case _: UnresolvedAlias =>
- case namedExpr: NamedExpression => validExpressionsMap.put(namedExpr.name, namedExpr)
+ case attr: AttributeReference =>
+ namedExpressionList += attr
+ case alias@Alias(_: AttributeReference, _) =>
+ namedExpressionList += alias
+ case alias@Alias(aggExp: AggregateExpression, name) =>
+ // get the updated expression for avg convert it to two expression
+ // sum and count
+ val expressions = PreAggregateUtil.validateAggregateFunctionAndGetFields(aggExp)
+ // if size is more than one then it was for average
+ if(expressions.size > 1) {
+ val sumExp = PreAggregateUtil.normalizeExprId(
+ expressions.head,
+ aggregate.allAttributes)
+ // get the logical plan fro count expression
+ val countExp = PreAggregateUtil.normalizeExprId(
+ expressions.last,
+ aggregate.allAttributes)
+ // check with same expression already sum is present then do not add to
+ // named expression list otherwise update the list and add it to set
+ if (!validExpressionsMap.contains(AggExpToColumnMappingModel(sumExp))) {
+ namedExpressionList +=
+ Alias(expressions.head, name + "_ sum")(NamedExpression.newExprId,
+ alias.qualifier,
+ Some(alias.metadata),
+ alias.isGenerated)
+ validExpressionsMap += AggExpToColumnMappingModel(sumExp)
+ }
+ // check with same expression already count is present then do not add to
+ // named expression list otherwise update the list and add it to set
+ if (!validExpressionsMap.contains(AggExpToColumnMappingModel(countExp))) {
+ namedExpressionList +=
+ Alias(expressions.last, name + "_ count")(NamedExpression.newExprId,
+ alias.qualifier,
+ Some(alias.metadata),
+ alias.isGenerated)
+ validExpressionsMap += AggExpToColumnMappingModel(countExp)
+ }
+ } else {
+ // get the logical plan for expression
+ val exp = PreAggregateUtil.normalizeExprId(
+ expressions.head,
+ aggregate.allAttributes)
+ // check with same expression already present then do not add to
+ // named expression list otherwise update the list and add it to set
+ if (!validExpressionsMap.contains(AggExpToColumnMappingModel(exp))) {
+ namedExpressionList+=alias
+ validExpressionsMap += AggExpToColumnMappingModel(exp)
+ }
+ }
+ case alias@Alias(_: Expression, _) =>
+ namedExpressionList += alias
}
- aggregate.copy(aggregateExpressions = validExpressionsMap.values.toSeq)
+ aggregate.copy(aggregateExpressions = namedExpressionList.toSeq)
case plan: LogicalPlan => plan
}
}
- /**
- * This method will split the avg column into sum and count and will return a sequence of tuple
- * of unique name, alias
- *
- */
- private def validateAggregateFunctionAndGetAlias(alias: Alias): Seq[(String,
- NamedExpression)] = {
- alias match {
- case udf@Alias(_: ScalaUDF, name) =>
- Seq((name, udf))
- case alias@Alias(attrExpression: AggregateExpression, _) =>
- attrExpression.aggregateFunction match {
- case Sum(attr: AttributeReference) =>
- (attr.name + "_sum", alias) :: Nil
- case Sum(MatchCastExpression(attr: AttributeReference, _)) =>
- (attr.name + "_sum", alias) :: Nil
- case Count(Seq(attr: AttributeReference)) =>
- (attr.name + "_count", alias) :: Nil
- case Count(Seq(MatchCastExpression(attr: AttributeReference, _))) =>
- (attr.name + "_count", alias) :: Nil
- case Average(attr: AttributeReference) =>
- Seq((attr.name + "_sum", Alias(attrExpression.
- copy(aggregateFunction = Sum(attr),
- resultId = NamedExpression.newExprId), attr.name + "_sum")()),
- (attr.name, Alias(attrExpression.
- copy(aggregateFunction = Count(attr),
- resultId = NamedExpression.newExprId), attr.name + "_count")()))
- case Average(cast@MatchCastExpression(attr: AttributeReference, _)) =>
- Seq((attr.name + "_sum", Alias(attrExpression.
- copy(aggregateFunction = Sum(cast),
- resultId = NamedExpression.newExprId),
- attr.name + "_sum")()),
- (attr.name, Alias(attrExpression.
- copy(aggregateFunction = Count(cast), resultId =
- NamedExpression.newExprId), attr.name + "_count")()))
- case _ => Seq(("", alias))
- }
-
- }
- }
-
/**
* Called by PreAggregateLoadingRules to validate if plan is valid for applying rules or not.
* If the plan has PreAggLoad i.e Loading UDF and does not have PreAgg i.e Query UDF then it is
* valid.
- *
- * @param namedExpression
- * @return
+ * @param namedExpression named expressions
+ * @return valid or not
*/
private def validateAggregateExpressions(namedExpression: Seq[NamedExpression]): Boolean = {
val filteredExpressions = namedExpression.filterNot(_.isInstanceOf[UnresolvedAlias])
http://git-wip-us.apache.org/repos/asf/carbondata/blob/c70e73f1/integration/spark2/src/main/spark2.1/CarbonSessionState.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/spark2.1/CarbonSessionState.scala b/integration/spark2/src/main/spark2.1/CarbonSessionState.scala
index 3d84dd3..a1fa382 100644
--- a/integration/spark2/src/main/spark2.1/CarbonSessionState.scala
+++ b/integration/spark2/src/main/spark2.1/CarbonSessionState.scala
@@ -262,7 +262,7 @@ class CarbonAnalyzer(catalog: SessionCatalog,
analyzer: Analyzer) extends Analyzer(catalog, conf) {
override def execute(plan: LogicalPlan): LogicalPlan = {
var logicalPlan = analyzer.execute(plan)
- logicalPlan = CarbonPreAggregateDataLoadingRules(logicalPlan)
+ logicalPlan = CarbonPreAggregateDataLoadingRules(sparkSession).apply(logicalPlan)
CarbonPreAggregateQueryRules(sparkSession).apply(logicalPlan)
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/c70e73f1/integration/spark2/src/main/spark2.2/CarbonSessionState.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/spark2.2/CarbonSessionState.scala b/integration/spark2/src/main/spark2.2/CarbonSessionState.scala
index f563007..5046541 100644
--- a/integration/spark2/src/main/spark2.2/CarbonSessionState.scala
+++ b/integration/spark2/src/main/spark2.2/CarbonSessionState.scala
@@ -190,7 +190,7 @@ class CarbonAnalyzer(catalog: SessionCatalog,
analyzer: Analyzer) extends Analyzer(catalog, conf) {
override def execute(plan: LogicalPlan): LogicalPlan = {
var logicalPlan = analyzer.execute(plan)
- logicalPlan = CarbonPreAggregateDataLoadingRules(logicalPlan)
+ logicalPlan = CarbonPreAggregateDataLoadingRules(sparkSession).apply(logicalPlan)
CarbonPreAggregateQueryRules(sparkSession).apply(logicalPlan)
}
}
[7/7] carbondata git commit: [CARBONDATA-2006] project level update
Posted by ja...@apache.org.
[CARBONDATA-2006] project level update
This closes #1776
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/90921eb7
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/90921eb7
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/90921eb7
Branch: refs/heads/carbonstore
Commit: 90921eb7bfe7e4e00902468b7073258b592f71ba
Parents: ed2c01f
Author: chenliang613 <ch...@huawei.com>
Authored: Mon Jan 8 18:54:22 2018 +0800
Committer: QiangCai <qi...@qq.com>
Committed: Mon Jan 8 19:40:05 2018 +0800
----------------------------------------------------------------------
NOTICE | 2 +-
build/README.md | 23 +++-----------------
.../execution/command/carbonTableSchema.scala | 0
integration/spark2/pom.xml | 2 +-
pom.xml | 8 +++----
5 files changed, 9 insertions(+), 26 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/90921eb7/NOTICE
----------------------------------------------------------------------
diff --git a/NOTICE b/NOTICE
index 3440816..fb63d08 100644
--- a/NOTICE
+++ b/NOTICE
@@ -1,5 +1,5 @@
Apache CarbonData
-Copyright 2016-2017 The Apache Software Foundation
+Copyright 2016 and onwards The Apache Software Foundation.
This product includes software developed at
The Apache Software Foundation (http://www.apache.org/).
http://git-wip-us.apache.org/repos/asf/carbondata/blob/90921eb7/build/README.md
----------------------------------------------------------------------
diff --git a/build/README.md b/build/README.md
index ab53877..65e58d8 100644
--- a/build/README.md
+++ b/build/README.md
@@ -25,31 +25,14 @@
* [Apache Thrift 0.9.3](http://archive.apache.org/dist/thrift/0.9.3/)
## Build command
-From 1.3.0 onwards, CarbonData supports spark 2.x, build without test,by default carbondata takes Spark 2.1.0 to build the project
+Build with different supported versions of Spark, by default using Spark 2.2.1 to build
```
-mvn -DskipTests clean package
-```
-
-Build with test
-```
-mvn clean package
-```
-
-Before 1.3.0, build with different supported versions of Spark
-```
-mvn -DskipTests -Pspark-1.5 -Dspark.version=1.5.1 clean package
-mvn -DskipTests -Pspark-1.5 -Dspark.version=1.5.2 clean package
-
-mvn -DskipTests -Pspark-1.6 -Dspark.version=1.6.1 clean package
-mvn -DskipTests -Pspark-1.6 -Dspark.version=1.6.2 clean package
-mvn -DskipTests -Pspark-1.6 -Dspark.version=1.6.3 clean package
-
mvn -DskipTests -Pspark-2.1 -Dspark.version=2.1.0 clean package
+mvn -DskipTests -Pspark-2.2 -Dspark.version=2.2.1 clean package
```
-
## For contributors : To build the format code after any changes, please follow the below command.
Note:Need install Apache Thrift 0.9.3
```
-mvn clean -DskipTests -Pbuild-with-format -Pspark-2.1 package
+mvn clean -DskipTests -Pbuild-with-format -Pspark-2.2 package
```
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/carbondata/blob/90921eb7/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
deleted file mode 100644
index e69de29..0000000
http://git-wip-us.apache.org/repos/asf/carbondata/blob/90921eb7/integration/spark2/pom.xml
----------------------------------------------------------------------
diff --git a/integration/spark2/pom.xml b/integration/spark2/pom.xml
index e470b23..472435e 100644
--- a/integration/spark2/pom.xml
+++ b/integration/spark2/pom.xml
@@ -154,7 +154,7 @@
<profile>
<id>build-all</id>
<properties>
- <spark.version>2.1.0</spark.version>
+ <spark.version>2.2.1</spark.version>
<scala.binary.version>2.11</scala.binary.version>
<scala.version>2.11.8</scala.version>
</properties>
http://git-wip-us.apache.org/repos/asf/carbondata/blob/90921eb7/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 6253285..0cbfc7b 100644
--- a/pom.xml
+++ b/pom.xml
@@ -428,7 +428,7 @@
<!-- This profile only should be used for release prepare to cover all the modules -->
<id>build-all</id>
<properties>
- <spark.version>2.1.0</spark.version>
+ <spark.version>2.2.1</spark.version>
<maven.test.skip>true</maven.test.skip>
</properties>
<modules>
@@ -460,9 +460,6 @@
</profile>
<profile>
<id>spark-2.1</id>
- <activation>
- <activeByDefault>true</activeByDefault>
- </activation>
<properties>
<spark.version>2.1.0</spark.version>
<scala.binary.version>2.11</scala.binary.version>
@@ -513,6 +510,9 @@
</profile>
<profile>
<id>spark-2.2</id>
+ <activation>
+ <activeByDefault>true</activeByDefault>
+ </activation>
<properties>
<spark.version>2.2.1</spark.version>
<scala.binary.version>2.11</scala.binary.version>
[3/7] carbondata git commit:
[CARBONDATA-1926][CARBONDATA-1927][Pre-Aggregate] Expression support inside
aggregate function for Query
Posted by ja...@apache.org.
[CARBONDATA-1926][CARBONDATA-1927][Pre-Aggregate] Expression support inside aggregate function for Query
Support transforming of query plan for aggregate table when query aggregate function contains any expression
Support sub query in Preaggregate table
This closes #1728
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/c70e73f1
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/c70e73f1
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/c70e73f1
Branch: refs/heads/carbonstore
Commit: c70e73f11b281cde9e8310d5a2d5640c75c30665
Parents: af4277e
Author: kumarvishal <ku...@gmail.com>
Authored: Mon Dec 25 15:04:39 2017 +0530
Committer: ravipesala <ra...@gmail.com>
Committed: Mon Jan 8 11:52:43 2018 +0530
----------------------------------------------------------------------
.../schema/table/AggregationDataMapSchema.java | 51 +-
.../core/preagg/AggregateQueryPlan.java | 48 +
.../core/preagg/AggregateTableSelector.java | 33 +-
.../carbondata/core/preagg/QueryColumn.java | 27 +-
.../carbondata/core/preagg/QueryPlan.java | 59 -
.../TestPreAggregateExpressions.scala | 68 +
.../TestPreAggregateTableSelection.scala | 5 +
.../TestPreAggregateWithSubQuery.scala | 88 +
.../preaaggregate/PreAggregateUtil.scala | 203 ++-
.../sql/hive/CarbonPreAggregateRules.scala | 1633 ++++++++----------
.../src/main/spark2.1/CarbonSessionState.scala | 2 +-
.../src/main/spark2.2/CarbonSessionState.scala | 2 +-
12 files changed, 1101 insertions(+), 1118 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/c70e73f1/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/AggregationDataMapSchema.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/AggregationDataMapSchema.java b/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/AggregationDataMapSchema.java
index 4b2d492..e061812 100644
--- a/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/AggregationDataMapSchema.java
+++ b/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/AggregationDataMapSchema.java
@@ -62,6 +62,8 @@ public class AggregationDataMapSchema extends DataMapSchema {
*/
private int ordinal = Integer.MAX_VALUE;
+ private Set aggExpToColumnMapping;
+
public AggregationDataMapSchema(String dataMapName, String className) {
super(dataMapName, className);
}
@@ -203,23 +205,6 @@ public class AggregationDataMapSchema extends DataMapSchema {
}
/**
- * Below method is to check if parent column with matching aggregate function
- * @param parentColumnName
- * parent column name
- * @param aggFunction
- * aggregate function
- * @return is matching
- */
- public boolean isColumnWithAggFunctionExists(String parentColumnName, String aggFunction) {
- Set<String> aggFunctions = parentColumnToAggregationsMapping.get(parentColumnName);
- if (null != aggFunctions && aggFunctions.contains(aggFunction)) {
- return true;
- }
- return false;
- }
-
-
- /**
* Method to prepare mapping of parent to list of aggregation function applied on that column
* @param listOfColumns
* child column schema list
@@ -341,4 +326,36 @@ public class AggregationDataMapSchema extends DataMapSchema {
public int getOrdinal() {
return ordinal;
}
+
+ /**
+ * Below method will be used to get the aggregation column based on index
+ * It will return the first aggregation column found based on index
+ * @param searchStartIndex
+ * start index
+ * @param sortedColumnSchema
+ * list of sorted table columns
+ * @return found column list
+ *
+ */
+ public ColumnSchema getAggColumnBasedOnIndex(int searchStartIndex,
+ List<ColumnSchema> sortedColumnSchema) {
+ ColumnSchema columnSchema = null;
+ for (int i = searchStartIndex; i < sortedColumnSchema.size(); i++) {
+ if (!sortedColumnSchema.get(i).getAggFunction().isEmpty()) {
+ columnSchema = sortedColumnSchema.get(i);
+ break;
+ }
+ }
+ return columnSchema;
+ }
+
+ public synchronized Set getAggExpToColumnMapping() {
+ return aggExpToColumnMapping;
+ }
+
+ public synchronized void setAggExpToColumnMapping(Set aggExpToColumnMapping) {
+ if (null == this.aggExpToColumnMapping) {
+ this.aggExpToColumnMapping = aggExpToColumnMapping;
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/c70e73f1/core/src/main/java/org/apache/carbondata/core/preagg/AggregateQueryPlan.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/preagg/AggregateQueryPlan.java b/core/src/main/java/org/apache/carbondata/core/preagg/AggregateQueryPlan.java
new file mode 100644
index 0000000..e895bde
--- /dev/null
+++ b/core/src/main/java/org/apache/carbondata/core/preagg/AggregateQueryPlan.java
@@ -0,0 +1,48 @@
+/*
+ * 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.core.preagg;
+
+import java.util.List;
+
+/**
+ * class to maintain the query plan to select the data map tables
+ */
+public class AggregateQueryPlan {
+
+ /**
+ * List of projection columns
+ */
+ private List<QueryColumn> projectionColumn;
+
+ /**
+ * list of filter columns
+ */
+ private List<QueryColumn> filterColumns;
+
+ public AggregateQueryPlan(List<QueryColumn> projectionColumn, List<QueryColumn> filterColumns) {
+ this.projectionColumn = projectionColumn;
+ this.filterColumns = filterColumns;
+ }
+
+ public List<QueryColumn> getProjectionColumn() {
+ return projectionColumn;
+ }
+
+ public List<QueryColumn> getFilterColumns() {
+ return filterColumns;
+ }
+}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/c70e73f1/core/src/main/java/org/apache/carbondata/core/preagg/AggregateTableSelector.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/preagg/AggregateTableSelector.java b/core/src/main/java/org/apache/carbondata/core/preagg/AggregateTableSelector.java
index 5347567..79d0904 100644
--- a/core/src/main/java/org/apache/carbondata/core/preagg/AggregateTableSelector.java
+++ b/core/src/main/java/org/apache/carbondata/core/preagg/AggregateTableSelector.java
@@ -36,15 +36,15 @@ public class AggregateTableSelector {
/**
* current query plan
*/
- private QueryPlan queryPlan;
+ private AggregateQueryPlan aggregateQueryPlan;
/**
* parent table
*/
private CarbonTable parentTable;
- public AggregateTableSelector(QueryPlan queryPlan, CarbonTable parentTable) {
- this.queryPlan = queryPlan;
+ public AggregateTableSelector(AggregateQueryPlan aggregateQueryPlan, CarbonTable parentTable) {
+ this.aggregateQueryPlan = aggregateQueryPlan;
this.parentTable = parentTable;
}
@@ -58,9 +58,8 @@ public class AggregateTableSelector {
* @return selected pre aggregate table schema
*/
public List<DataMapSchema> selectPreAggDataMapSchema() {
- List<QueryColumn> projectionColumn = queryPlan.getProjectionColumn();
- List<QueryColumn> aggColumns = queryPlan.getAggregationColumns();
- List<QueryColumn> filterColumns = queryPlan.getFilterColumns();
+ List<QueryColumn> projectionColumn = aggregateQueryPlan.getProjectionColumn();
+ List<QueryColumn> filterColumns = aggregateQueryPlan.getFilterColumns();
List<DataMapSchema> dataMapSchemaList = parentTable.getTableInfo().getDataMapSchemaList();
List<DataMapSchema> selectedDataMapSchema = new ArrayList<>();
boolean isMatch;
@@ -74,6 +73,7 @@ public class AggregateTableSelector {
getColumnSchema(queryColumn, aggregationDataMapSchema);
if (null == columnSchemaByParentName) {
isMatch = false;
+ break;
}
}
if (isMatch) {
@@ -99,6 +99,7 @@ public class AggregateTableSelector {
getColumnSchema(queryColumn, aggregationDataMapSchema);
if (null == columnSchemaByParentName) {
isMatch = false;
+ break;
}
}
if (isMatch) {
@@ -110,26 +111,6 @@ public class AggregateTableSelector {
return selectedDataMapSchema;
}
}
- // match aggregation columns
- if (null != aggColumns && !aggColumns.isEmpty()) {
- List<DataMapSchema> dmSchemaToIterate =
- selectedDataMapSchema.isEmpty() ? dataMapSchemaList : selectedDataMapSchema;
- selectedDataMapSchema = new ArrayList<>();
- for (DataMapSchema dmSchema : dmSchemaToIterate) {
- isMatch = true;
- for (QueryColumn queryColumn : aggColumns) {
- AggregationDataMapSchema aggregationDataMapSchema = (AggregationDataMapSchema) dmSchema;
- if (!aggregationDataMapSchema
- .isColumnWithAggFunctionExists(queryColumn.getColumnSchema().getColumnName(),
- queryColumn.getAggFunction())) {
- isMatch = false;
- }
- }
- if (isMatch) {
- selectedDataMapSchema.add(dmSchema);
- }
- }
- }
return selectedDataMapSchema;
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/c70e73f1/core/src/main/java/org/apache/carbondata/core/preagg/QueryColumn.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/preagg/QueryColumn.java b/core/src/main/java/org/apache/carbondata/core/preagg/QueryColumn.java
index c91a703..42daa6c 100644
--- a/core/src/main/java/org/apache/carbondata/core/preagg/QueryColumn.java
+++ b/core/src/main/java/org/apache/carbondata/core/preagg/QueryColumn.java
@@ -30,16 +30,6 @@ public class QueryColumn {
private ColumnSchema columnSchema;
/**
- * to store the change data type in case of cast
- */
- private String changedDataType;
-
- /**
- * aggregation function applied
- */
- private String aggFunction;
-
- /**
* is filter column
*/
private boolean isFilterColumn;
@@ -49,11 +39,8 @@ public class QueryColumn {
*/
private String timeseriesFunction;
- public QueryColumn(ColumnSchema columnSchema, String changedDataType, String aggFunction,
- boolean isFilterColumn, String timeseriesFunction) {
+ public QueryColumn(ColumnSchema columnSchema, boolean isFilterColumn, String timeseriesFunction) {
this.columnSchema = columnSchema;
- this.changedDataType = changedDataType;
- this.aggFunction = aggFunction;
this.isFilterColumn = isFilterColumn;
this.timeseriesFunction = timeseriesFunction;
}
@@ -62,14 +49,6 @@ public class QueryColumn {
return columnSchema;
}
- public String getChangedDataType() {
- return changedDataType;
- }
-
- public String getAggFunction() {
- return aggFunction;
- }
-
public boolean isFilterColumn() {
return isFilterColumn;
}
@@ -92,9 +71,6 @@ public class QueryColumn {
if (!columnSchema.equals(that.columnSchema)) {
return false;
}
- if (!(aggFunction != null ? aggFunction.equals(that.aggFunction) : that.aggFunction == null)) {
- return false;
- }
return timeseriesFunction != null ?
timeseriesFunction.equals(that.timeseriesFunction) :
that.timeseriesFunction == null;
@@ -102,7 +78,6 @@ public class QueryColumn {
@Override public int hashCode() {
int result = columnSchema.hashCode();
- result = 31 * result + (aggFunction != null ? aggFunction.hashCode() : 0);
result = 31 * result + (timeseriesFunction != null ? timeseriesFunction.hashCode() : 0);
result = 31 * result + (isFilterColumn ? 1 : 0);
return result;
http://git-wip-us.apache.org/repos/asf/carbondata/blob/c70e73f1/core/src/main/java/org/apache/carbondata/core/preagg/QueryPlan.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/preagg/QueryPlan.java b/core/src/main/java/org/apache/carbondata/core/preagg/QueryPlan.java
deleted file mode 100644
index 21a34fa..0000000
--- a/core/src/main/java/org/apache/carbondata/core/preagg/QueryPlan.java
+++ /dev/null
@@ -1,59 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.carbondata.core.preagg;
-
-import java.util.List;
-
-/**
- * class to maintain the query plan to select the data map tables
- */
-public class QueryPlan {
-
- /**
- * List of projection columns
- */
- private List<QueryColumn> projectionColumn;
-
- /**
- * list of aggregation columns
- */
- private List<QueryColumn> aggregationColumns;
-
- /**
- * list of filter columns
- */
- private List<QueryColumn> filterColumns;
-
- public QueryPlan(List<QueryColumn> projectionColumn, List<QueryColumn> aggregationColumns,
- List<QueryColumn> filterColumns) {
- this.projectionColumn = projectionColumn;
- this.aggregationColumns = aggregationColumns;
- this.filterColumns = filterColumns;
- }
-
- public List<QueryColumn> getProjectionColumn() {
- return projectionColumn;
- }
-
- public List<QueryColumn> getAggregationColumns() {
- return aggregationColumns;
- }
-
- public List<QueryColumn> getFilterColumns() {
- return filterColumns;
- }
-}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/c70e73f1/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateExpressions.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateExpressions.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateExpressions.scala
index 4171690..0b22c56 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateExpressions.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateExpressions.scala
@@ -17,6 +17,10 @@
package org.apache.carbondata.integration.spark.testsuite.preaggregate
+import org.apache.spark.sql.{CarbonDatasourceHadoopRelation, Row}
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.execution.datasources.LogicalRelation
+import org.apache.spark.sql.hive.CarbonRelation
import org.apache.spark.sql.test.util.QueryTest
import org.scalatest.BeforeAndAfterAll
@@ -25,6 +29,7 @@ class TestPreAggregateExpressions extends QueryTest with BeforeAndAfterAll {
override def beforeAll: Unit = {
sql("DROP TABLE IF EXISTS mainTable")
sql("CREATE TABLE mainTable(id int, name string, city string, age string) STORED BY 'org.apache.carbondata.format'")
+ sql(s"LOAD DATA LOCAL INPATH '$resourcesPath/measureinsertintotest.csv' into table mainTable")
}
test("test pre agg create table with expression 1") {
sql(
@@ -94,6 +99,69 @@ class TestPreAggregateExpressions extends QueryTest with BeforeAndAfterAll {
| """.stripMargin)
checkExistence(sql("DESCRIBE FORMATTED mainTable_agg5"), true, "maintable_column_0_count")
}
+ test("test pre agg table selection with expression 1") {
+ val df = sql("select name as NewName, count(age) as sum from mainTable group by name order by name")
+ preAggTableValidator(df.queryExecution.analyzed, "maintable_agg0")
+ }
+
+
+ test("test pre agg table selection with expression 2") {
+ val df = sql("select name as NewName, sum(case when age=35 then id else 0 end) as sum from mainTable group by name order by name")
+ preAggTableValidator(df.queryExecution.analyzed, "maintable_agg1")
+ }
+
+ test("test pre agg table selection with expression 3") {
+ val df = sql("select sum(case when age=35 then id else 0 end) from maintable")
+ checkAnswer(df, Seq(Row(6.0)))
+ }
+
+ test("test pre agg table selection with expression 4") {
+ val df = sql("select sum(case when age=27 then id else 0 end) from maintable")
+ preAggTableValidator(df.queryExecution.analyzed, "maintable_agg3")
+ checkAnswer(df, Seq(Row(2.0)))
+ }
+
+ test("test pre agg table selection with expression 5") {
+ val df = sql("select sum(case when age=27 then id else 0 end), sum(case when age=35 then id else 0 end) from maintable")
+ preAggTableValidator(df.queryExecution.analyzed, "maintable_agg4")
+ checkAnswer(df, Seq(Row(2.0,6.0)))
+ }
+
+ /**
+ * Below method will be used to validate the table name is present in the plan or not
+ * @param plan
+ * query plan
+ * @param actualTableName
+ * table name to be validated
+ */
+ def preAggTableValidator(plan: LogicalPlan, actualTableName: String) : Unit ={
+ var isValidPlan = false
+ plan.transform {
+ // first check if any preaTable1 scala function is applied it is present is in plan
+ // then call is from create preaTable1regate table class so no need to transform the query plan
+ case ca:CarbonRelation =>
+ if (ca.isInstanceOf[CarbonDatasourceHadoopRelation]) {
+ val relation = ca.asInstanceOf[CarbonDatasourceHadoopRelation]
+ if(relation.carbonTable.getTableName.equalsIgnoreCase(actualTableName)) {
+ isValidPlan = true
+ }
+ }
+ ca
+ case logicalRelation:LogicalRelation =>
+ if(logicalRelation.relation.isInstanceOf[CarbonDatasourceHadoopRelation]) {
+ val relation = logicalRelation.relation.asInstanceOf[CarbonDatasourceHadoopRelation]
+ if(relation.carbonTable.getTableName.equalsIgnoreCase(actualTableName)) {
+ isValidPlan = true
+ }
+ }
+ logicalRelation
+ }
+ if(!isValidPlan) {
+ assert(false)
+ } else {
+ assert(true)
+ }
+ }
override def afterAll: Unit = {
sql("DROP TABLE IF EXISTS mainTable")
http://git-wip-us.apache.org/repos/asf/carbondata/blob/c70e73f1/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateTableSelection.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateTableSelection.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateTableSelection.scala
index fdeb2bd..322827e 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateTableSelection.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateTableSelection.scala
@@ -230,6 +230,11 @@ class TestPreAggregateTableSelection extends QueryTest with BeforeAndAfterAll {
checkAnswer(df, Row("vishal", 29))
}
+ test("test PreAggregate table selection 29") {
+ val df = sql("select sum(id) from mainTable group by name")
+ preAggTableValidator(df.queryExecution.analyzed, "maintable_agg2")
+ }
+
override def afterAll: Unit = {
sql("drop table if exists mainTable")
sql("drop table if exists lineitem")
http://git-wip-us.apache.org/repos/asf/carbondata/blob/c70e73f1/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateWithSubQuery.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateWithSubQuery.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateWithSubQuery.scala
new file mode 100644
index 0000000..0f87803
--- /dev/null
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateWithSubQuery.scala
@@ -0,0 +1,88 @@
+/*
+ * 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.integration.spark.testsuite.preaggregate
+
+import org.apache.spark.sql.CarbonDatasourceHadoopRelation
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.execution.datasources.LogicalRelation
+import org.apache.spark.sql.test.util.QueryTest
+import org.scalatest.BeforeAndAfterAll
+
+
+class TestPreAggregateWithSubQuery extends QueryTest with BeforeAndAfterAll {
+
+ override def beforeAll: Unit = {
+ sql("drop table if exists mainTable")
+ sql("CREATE TABLE mainTable(id int, name string, city string, age string) STORED BY 'org.apache.carbondata.format'")
+ sql("CREATE TABLE mainTable1(id int, name string, city string, age string) STORED BY 'org.apache.carbondata.format'")
+ sql("create datamap agg0 on table mainTable using 'preaggregate' as select name,sum(age) from mainTable group by name")
+ sql(s"LOAD DATA LOCAL INPATH '$resourcesPath/measureinsertintotest.csv' into table mainTable")
+ sql(s"LOAD DATA LOCAL INPATH '$resourcesPath/measureinsertintotest.csv' into table mainTable1")
+ }
+
+ test("test sub query PreAggregate table selection 1") {
+ val df = sql("select t2.newnewname as newname from mainTable1 t1 join (select name as newnewname,sum(age) as sum from mainTable group by name )t2 on t1.name=t2.newnewname group by t2.newnewname")
+ matchTable(collectLogicalRelation(df.queryExecution.analyzed), "maintable_agg0")
+ }
+
+ test("test sub query PreAggregate table selection 2") {
+ val df = sql("select t1.name,t1.city from mainTable1 t1 join (select name as newnewname,sum(age) as sum from mainTable group by name )t2 on t1.name=t2.newnewname")
+ matchTable(collectLogicalRelation(df.queryExecution.analyzed), "maintable_agg0")
+ }
+
+ test("test sub query PreAggregate table selection 3") {
+ val df = sql("select t1.name,t2.sum from mainTable1 t1 join (select name as newnewname,sum(age) as sum from mainTable group by name )t2 on t1.name=t2.newnewname")
+ matchTable(collectLogicalRelation(df.queryExecution.analyzed), "maintable_agg0")
+ }
+
+ test("test sub query PreAggregate table selection 4") {
+ val df = sql("select t1.name,t2.sum from mainTable1 t1 join (select name,sum(age) as sum from mainTable group by name )t2 on t1.name=t2.name group by t1.name, t2.sum")
+ matchTable(collectLogicalRelation(df.queryExecution.analyzed), "maintable_agg0")
+ }
+
+ /**
+ * Below method will be used to collect all the logical relation from logical plan
+ * @param logicalPlan
+ * query logical plan
+ * @return all the logical relation
+ */
+ def collectLogicalRelation(logicalPlan: LogicalPlan) : Seq[LogicalRelation] = {
+ logicalPlan.collect{
+ case l:LogicalRelation => l
+ }
+ }
+
+ /**
+ * Below method will be used to match the logical relation
+ * @param logicalRelations
+ * all logical relation
+ * @param tableName
+ * table name
+ */
+ def matchTable(logicalRelations: Seq[LogicalRelation], tableName: String) {
+ assert(logicalRelations.exists {
+ case l:LogicalRelation if l.relation.isInstanceOf[CarbonDatasourceHadoopRelation] =>
+ l.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonTable.getTableName.
+ equalsIgnoreCase(tableName)
+ })
+ }
+
+ override def afterAll: Unit = {
+ sql("drop table if exists mainTable")
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/c70e73f1/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateUtil.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateUtil.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateUtil.scala
index 217436d..86d0c6a 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateUtil.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateUtil.scala
@@ -20,18 +20,18 @@ import scala.collection.mutable.{ArrayBuffer, ListBuffer}
import scala.collection.JavaConverters._
import org.apache.spark.sql.{CarbonDatasourceHadoopRelation, CarbonEnv, CarbonSession, SparkSession}
-import org.apache.spark.sql.CarbonExpressions.{CarbonSubqueryAlias => SubqueryAlias, MatchCast => Cast, MatchCastExpression}
+import org.apache.spark.sql.CarbonExpressions.{CarbonSubqueryAlias => SubqueryAlias, MatchCastExpression}
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.analysis.{UnresolvedAlias, UnresolvedFunction, UnresolvedRelation}
-import org.apache.spark.sql.catalyst.expressions.{Alias, AttributeReference, Expression, NamedExpression, ScalaUDF}
-import org.apache.spark.sql.catalyst.expressions.aggregate._
+import org.apache.spark.sql.catalyst.expressions.{Alias, AttributeReference, AttributeSeq, Cast, Expression, ExprId, NamedExpression, ScalaUDF}
+import org.apache.spark.sql.catalyst.expressions.aggregate.{Count, _}
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.execution.command.{ColumnTableRelation, DataMapField, Field}
import org.apache.spark.sql.execution.command.management.CarbonLoadDataCommand
import org.apache.spark.sql.execution.datasources.LogicalRelation
import org.apache.spark.sql.hive.CarbonRelation
import org.apache.spark.sql.parser.CarbonSpark2SqlParser
-import org.apache.spark.sql.types.DataType
+import org.apache.spark.sql.types.{DataType, LongType}
import org.apache.carbondata.common.logging.{LogService, LogServiceFactory}
import org.apache.carbondata.core.constants.CarbonCommonConstants
@@ -85,14 +85,10 @@ object PreAggregateUtil {
/**
* Below method will be used to get the fields from expressions
- * @param groupByExp
- * grouping expression
- * @param aggExp
- * aggregate expression
- * @param logicalRelation
- * logical relation
- * @param selectStmt
- * select statement
+ * @param groupByExp grouping expression
+ * @param aggExp aggregate expression
+ * @param logicalRelation logical relation
+ * @param selectStmt select statement
* @return fields from expressions
*/
def getFieldsFromPlan(groupByExp: Seq[Expression],
@@ -165,16 +161,11 @@ object PreAggregateUtil {
/**
* Below method will be used to get the column relation
* with the parent column which will be used during query and data loading
- * @param parentColumnName
- * parent column name
- * @param parentTableId
- * parent column id
- * @param parentTableName
- * parent table name
- * @param parentDatabaseName
- * parent database name
- * @param carbonTable
- * carbon table
+ * @param parentColumnName parent column name
+ * @param parentTableId parent column id
+ * @param parentTableName parent table name
+ * @param parentDatabaseName parent database name
+ * @param carbonTable carbon table
* @return column relation object
*/
def getColumnRelation(parentColumnName: String,
@@ -280,7 +271,7 @@ object PreAggregateUtil {
"sum")
list += createFieldForAggregateExpression(
exp,
- changeDataType,
+ LongType,
carbonTable,
newColumnName,
"count")
@@ -293,7 +284,7 @@ object PreAggregateUtil {
"sum")
list += createFieldForAggregateExpression(
exp,
- avg.dataType,
+ LongType,
carbonTable,
newColumnName,
"count")
@@ -306,16 +297,11 @@ object PreAggregateUtil {
/**
* Below method will be used to get the field and its data map field object
* for aggregate expression
- * @param expression
- * expression in aggregate function
- * @param dataType
- * data type
- * @param carbonTable
- * parent carbon table
- * @param newColumnName
- * column name of aggregate table
- * @param aggregationName
- * aggregate function name
+ * @param expression expression in aggregate function
+ * @param dataType data type
+ * @param carbonTable parent carbon table
+ * @param newColumnName column name of aggregate table
+ * @param aggregationName aggregate function name
* @return field and its metadata tuple
*/
def createFieldForAggregateExpression(
@@ -552,8 +538,7 @@ object PreAggregateUtil {
* Below method will be used to update logical plan
* this is required for creating pre aggregate tables,
* so @CarbonPreAggregateRules will not be applied during creation
- * @param logicalPlan
- * actual logical plan
+ * @param logicalPlan actual logical plan
* @return updated plan
*/
def updatePreAggQueyPlan(logicalPlan: LogicalPlan): LogicalPlan = {
@@ -654,10 +639,8 @@ object PreAggregateUtil {
/**
* Below method will be used to get the select query when rollup policy is
* applied in case of timeseries table
- * @param tableSchema
- * main data map schema
- * @param selectedDataMapSchema
- * selected data map schema for rollup
+ * @param tableSchema main data map schema
+ * @param selectedDataMapSchema selected data map schema for rollup
* @return select query based on rolloup
*/
def createTimeseriesSelectQueryForRollup(
@@ -695,10 +678,8 @@ object PreAggregateUtil {
* Below method will be used to creating select query for timeseries
* for lowest level for aggergation like second level, in that case it will
* hit the maintable
- * @param tableSchema
- * data map schema
- * @param parentTableName
- * parent schema
+ * @param tableSchema data map schema
+ * @param parentTableName parent schema
* @return select query for loading
*/
def createTimeSeriesSelectQueryFromMain(tableSchema: TableSchema,
@@ -728,10 +709,8 @@ object PreAggregateUtil {
/**
* Below method will be used to select rollup table in case of
* timeseries data map loading
- * @param list
- * list of timeseries datamap
- * @param dataMapSchema
- * datamap schema
+ * @param list list of timeseries datamap
+ * @param dataMapSchema datamap schema
* @return select table name
*/
def getRollupDataMapNameForTimeSeries(
@@ -748,4 +727,132 @@ object PreAggregateUtil {
rollupDataMapSchema.lastOption
}
}
+
+ /**
+ * Below method will be used to validate aggregate function and get the attribute information
+ * which is applied on select query.
+ * Currently sum, max, min, count, avg is supported
+ * in case of any other aggregate function it will return empty sequence
+ * In case of avg it will return two fields one for count
+ * and other of sum of that column to support rollup
+ *
+ * @param aggExp aggregate expression
+ * @return list of fields
+ */
+ def validateAggregateFunctionAndGetFields(aggExp: AggregateExpression):
+ Seq[AggregateExpression] = {
+ aggExp.aggregateFunction match {
+ case Sum(MatchCastExpression(exp: Expression, changeDataType: DataType)) =>
+ Seq(AggregateExpression(Sum(Cast(
+ exp,
+ changeDataType)),
+ aggExp.mode,
+ aggExp.isDistinct))
+ case Sum(_: Expression) =>
+ Seq(aggExp)
+ case Count(MatchCastExpression(exp: Seq[Expression], changeDataType: DataType)) =>
+ Seq(AggregateExpression(Count(Cast(
+ exp,
+ changeDataType)),
+ aggExp.mode,
+ aggExp.isDistinct))
+ case Count(_: Seq[Expression]) =>
+ Seq(aggExp)
+ case Min(MatchCastExpression(exp: Expression, changeDataType: DataType)) =>
+ Seq(AggregateExpression(Min(Cast(
+ exp,
+ changeDataType)),
+ aggExp.mode,
+ aggExp.isDistinct))
+ case Min(exp: Expression) =>
+ Seq(aggExp)
+ case Max(MatchCastExpression(exp: Expression, changeDataType: DataType)) =>
+ Seq(AggregateExpression(Max(Cast(
+ exp,
+ changeDataType)),
+ aggExp.mode,
+ aggExp.isDistinct))
+ case Max(exp: Expression) =>
+ Seq(aggExp)
+ // in case of average need to return two columns
+ // sum and count of the column to added during table creation to support rollup
+ case Average(MatchCastExpression(exp: Expression, changeDataType: DataType)) =>
+ Seq(AggregateExpression(Sum(Cast(
+ exp,
+ changeDataType)),
+ aggExp.mode,
+ aggExp.isDistinct),
+ AggregateExpression(Count(Cast(
+ exp,
+ changeDataType)),
+ aggExp.mode,
+ aggExp.isDistinct))
+ // in case of average need to return two columns
+ // sum and count of the column to added during table creation to support rollup
+ case Average(exp: Expression) =>
+ Seq(AggregateExpression(Sum(exp),
+ aggExp.mode,
+ aggExp.isDistinct),
+ AggregateExpression(Count(exp),
+ aggExp.mode,
+ aggExp.isDistinct))
+ case _ =>
+ Seq.empty
+ }
+ }
+
+ /**
+ * Below method will be used to get the logical plan from aggregate expression
+ * @param aggExp aggregate expression
+ * @param tableName parent table name
+ * @param databaseName database name
+ * @param logicalRelation logical relation
+ * @return logical plan
+ */
+ def getLogicalPlanFromAggExp(aggExp: AggregateExpression,
+ tableName: String,
+ databaseName: String,
+ logicalRelation: LogicalRelation,
+ sparkSession: SparkSession,
+ parser: CarbonSpark2SqlParser): LogicalPlan = {
+ // adding the preAGG UDF, so pre aggregate data loading rule and query rule will not
+ // be applied
+ val query = parser.addPreAggFunction(s"Select ${ aggExp.sql } from $databaseName.$tableName")
+ // updating the logical relation of logical plan to so when two logical plan
+ // will be compared it will not consider relation
+ updateLogicalRelation(sparkSession.sql(query).logicalPlan, logicalRelation)
+ }
+
+ /**
+ * Below method will be used to update the logical plan of expression
+ * with parent table logical relation
+ * @param logicalPlan logial plan
+ * @param logicalRelation maintable logical relation
+ * @return updated plan
+ */
+ def updateLogicalRelation(logicalPlan: LogicalPlan,
+ logicalRelation: LogicalRelation): LogicalPlan = {
+ logicalPlan transform {
+ case l: LogicalRelation =>
+ l.copy(relation = logicalRelation.relation)
+ }
+ }
+
+ /**
+ * Normalize the exprIds in the given expression, by updating the exprId in `AttributeReference`
+ * with its referenced ordinal from input attributes. It's similar to `BindReferences` but we
+ * do not use `BindReferences` here as the plan may take the expression as a parameter with type
+ * `Attribute`, and replace it with `BoundReference` will cause error.
+ */
+ def normalizeExprId[T <: Expression](e: T, input: AttributeSeq): T = {
+ e.transformUp {
+ case ar: AttributeReference =>
+ val ordinal = input.indexOf(ar.exprId)
+ if (ordinal == -1) {
+ ar
+ } else {
+ ar.withExprId(ExprId(ordinal))
+ }
+ }.canonicalized.asInstanceOf[T]
+ }
}
[4/7] carbondata git commit: [CARBONDATA-1702] [Segment-Reading]
added doc for query from specified segments feature
Posted by ja...@apache.org.
[CARBONDATA-1702] [Segment-Reading] added doc for query from specified segments feature
add doc for query from specified segments
This closes #1727
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/705b1111
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/705b1111
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/705b1111
Branch: refs/heads/carbonstore
Commit: 705b1111de1e84f1bebb86b6550fcfed477e6d63
Parents: c70e73f
Author: praveenmeenakshi56 <pr...@gmail.com>
Authored: Wed Dec 27 19:30:52 2017 +0530
Committer: chenliang613 <ch...@huawei.com>
Committed: Mon Jan 8 15:11:05 2018 +0800
----------------------------------------------------------------------
docs/data-management-on-carbondata.md | 79 ++++++++++++++++++++++++++++++
1 file changed, 79 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/705b1111/docs/data-management-on-carbondata.md
----------------------------------------------------------------------
diff --git a/docs/data-management-on-carbondata.md b/docs/data-management-on-carbondata.md
index 20c70e3..39acfbf 100644
--- a/docs/data-management-on-carbondata.md
+++ b/docs/data-management-on-carbondata.md
@@ -781,3 +781,82 @@ This tutorial is going to introduce all commands and data operations on CarbonDa
```
DELETE FROM TABLE CarbonDatabase.CarbonTable WHERE SEGMENT.STARTTIME BEFORE '2017-06-01 12:05:06'
```
+### SEGMENT READING
+
+ This command is used to read data from specified segments during CarbonScan.
+
+
+ Get the Segment ID:
+
+ ```
+ SHOW SEGMENTS FOR TABLE [db_name.]table_name LIMIT number_of_segments
+ ```
+
+ Set the segment IDs
+
+ ```
+ SET cabon.input.segments.<database_name>.<table_name> = <list of segment IDs>;
+ ```
+
+ **Property:**
+
+ cabon.input.segments: Specifies the segment IDs to be queried. This property allows you to query specified segments of the specified table. The CarbonScan will read data from specified segments only.
+
+ ```
+ SET cabon.input.segments.<database_name>.<table_name> = <list of segment IDs>;
+ ```
+
+ If user wants to query with segments reading in multi threading mode, then CarbonSession.threadSet can be used instead of SET query.
+
+ ```
+ CarbonSession.threadSet ("cabon.input.segments.<database_name>.<table_name>","<list of segment IDs>");
+ ```
+
+ Reset the segment IDs:
+
+ ```
+ SET cabon.input.segments.<database_name>.<table_name> = *;
+ ```
+
+ If user wants to query with segments reading in multi threading mode, then CarbonSession.threadSet can be used instead of SET query.
+
+ ```
+ CarbonSession.threadSet ("cabon.input.segments.<database_name>.<table_name>","*");
+ ```
+
+ Reset
+
+ It will reset all the properties set for carbondata. It is not recommended if you do not want to reset all the properties except cabon.input.segments.
+
+ ```
+ RESET
+ ```
+
+ **NOTE**: It is not recommended to set this property in carbon.properties file, because all the sessions will take this segments list unless it is overwritten at session or thread level.
+
+ **Examples:**
+
+ * Example to show the list of segment IDs,segment status, and other required details and then specify the list of segments to be read.
+
+ ```
+ SHOW SEGMENTS FOR carbontable1;
+
+ SET cabon.input.segments.db.carbontable1 = 1,3,9;
+ ```
+
+ * Example to query with segments reading in multi threading mode:
+
+ ```
+ CarbonSession.threadSet ("cabon.input.segments.db.carbontable_Multi_Thread","1,3");
+ ```
+
+ * Example for threadset in multithread environment (following shows how it is used in Scala code):
+
+ ```
+ def main(args: Array[String]) {
+ Future {
+ CarbonSession.threadSet ("cabon.input.segments.db.carbontable_Multi_Thread","1")
+ spark.sql("select count(empno) from cabon.input.segments.db.carbontable_Multi_Thread").show();
+ }
+ }
+ ```
[6/7] carbondata git commit: [HOTFIX] Fix exception handling in
PartitionMapFileStore
Posted by ja...@apache.org.
[HOTFIX] Fix exception handling in PartitionMapFileStore
Exception should not be ignored in PartitionMapFileStore.java, it should be thrown instead.
This closes #1771
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/ed2c01f5
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/ed2c01f5
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/ed2c01f5
Branch: refs/heads/carbonstore
Commit: ed2c01f5d4d2da07449046271933f47e76b430af
Parents: 11353e2
Author: Jacky Li <ja...@qq.com>
Authored: Sun Jan 7 10:58:54 2018 +0800
Committer: ravipesala <ra...@gmail.com>
Committed: Mon Jan 8 15:24:52 2018 +0530
----------------------------------------------------------------------
.../carbondata/core/metadata/PartitionMapFileStore.java | 10 +++++-----
1 file changed, 5 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/ed2c01f5/core/src/main/java/org/apache/carbondata/core/metadata/PartitionMapFileStore.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/metadata/PartitionMapFileStore.java b/core/src/main/java/org/apache/carbondata/core/metadata/PartitionMapFileStore.java
index f7074c4..3068cd9 100644
--- a/core/src/main/java/org/apache/carbondata/core/metadata/PartitionMapFileStore.java
+++ b/core/src/main/java/org/apache/carbondata/core/metadata/PartitionMapFileStore.java
@@ -188,7 +188,7 @@ public class PartitionMapFileStore {
* @param partitionMapPath
* @return
*/
- private PartitionMapper readPartitionMap(String partitionMapPath) {
+ private PartitionMapper readPartitionMap(String partitionMapPath) throws IOException {
Gson gsonObjectToRead = new Gson();
DataInputStream dataInputStream = null;
BufferedReader buffReader = null;
@@ -206,10 +206,10 @@ public class PartitionMapFileStore {
Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET));
buffReader = new BufferedReader(inStream);
partitionMapper = gsonObjectToRead.fromJson(buffReader, PartitionMapper.class);
- } catch (IOException e) {
- return null;
} finally {
- CarbonUtil.closeStreams(buffReader, inStream, dataInputStream);
+ if (inStream != null) {
+ CarbonUtil.closeStreams(buffReader, inStream, dataInputStream);
+ }
}
return partitionMapper;
@@ -219,7 +219,7 @@ public class PartitionMapFileStore {
* Reads all partitions which existed inside the passed segment path
* @param segmentPath
*/
- public void readAllPartitionsOfSegment(String segmentPath) {
+ public void readAllPartitionsOfSegment(String segmentPath) throws IOException {
String partitionFilePath = getPartitionFilePath(segmentPath);
if (partitionFilePath != null) {
partionedSegment = true;