You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ch...@apache.org on 2016/12/29 02:04:18 UTC
[1/2] incubator-carbondata git commit: fix statistics code and adapt
query stastistics to spark2
Repository: incubator-carbondata
Updated Branches:
refs/heads/master dc7c86ef3 -> 629061816
fix statistics code and adapt query stastistics to spark2
fix style
add non filter blocket num
Project: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/commit/be6bb16d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/tree/be6bb16d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/diff/be6bb16d
Branch: refs/heads/master
Commit: be6bb16d5f31ba8376c3c15da39752d663ad35d6
Parents: dc7c86e
Author: eason <ea...@easondeMacBook-Pro.local>
Authored: Sat Dec 24 17:36:53 2016 +0800
Committer: chenliang613 <ch...@apache.org>
Committed: Thu Dec 29 10:02:48 2016 +0800
----------------------------------------------------------------------
.../QueryStatisticsRecorderImpl.java | 2 +-
.../processor/AbstractDataBlockIterator.java | 2 +-
.../scan/scanner/AbstractBlockletScanner.java | 17 +++++++++++++++
.../scan/scanner/impl/NonFilterScanner.java | 5 ++++-
.../QueryStasticsRecorderImplTest.java | 4 ++--
.../scanner/impl/FilterScannerTest.java | 2 +-
.../carbondata/spark/rdd/CarbonScanRDD.scala | 22 ++++++++++----------
.../apache/spark/sql/hive/CarbonMetastore.scala | 2 +-
8 files changed, 38 insertions(+), 18 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/be6bb16d/core/src/main/java/org/apache/carbondata/core/carbon/querystatistics/QueryStatisticsRecorderImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/carbon/querystatistics/QueryStatisticsRecorderImpl.java b/core/src/main/java/org/apache/carbondata/core/carbon/querystatistics/QueryStatisticsRecorderImpl.java
index bcf1b44..996e7f0 100644
--- a/core/src/main/java/org/apache/carbondata/core/carbon/querystatistics/QueryStatisticsRecorderImpl.java
+++ b/core/src/main/java/org/apache/carbondata/core/carbon/querystatistics/QueryStatisticsRecorderImpl.java
@@ -130,7 +130,7 @@ public class QueryStatisticsRecorderImpl implements QueryStatisticsRecorder,Seri
}
}
String headers = "task_id,load_blocks_time,load_dictionary_time,scan_blocks_time," +
- "total_executor_time,scan_blocks_num,total_blocklet" +
+ "total_executor_time,scan_blocks_num,total_blocklet," +
"valid_scan_blocklet,result_size";
List<String> values = new ArrayList<String>();
values.add(queryIWthTask);
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/be6bb16d/core/src/main/java/org/apache/carbondata/scan/processor/AbstractDataBlockIterator.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/scan/processor/AbstractDataBlockIterator.java b/core/src/main/java/org/apache/carbondata/scan/processor/AbstractDataBlockIterator.java
index ff3b31a..95d5836 100644
--- a/core/src/main/java/org/apache/carbondata/scan/processor/AbstractDataBlockIterator.java
+++ b/core/src/main/java/org/apache/carbondata/scan/processor/AbstractDataBlockIterator.java
@@ -93,7 +93,7 @@ public abstract class AbstractDataBlockIterator extends CarbonIterator<List<Obje
if (blockExecutionInfo.getFilterExecuterTree() != null) {
blockletScanner = new FilterScanner(blockExecutionInfo, queryStatisticsModel);
} else {
- blockletScanner = new NonFilterScanner(blockExecutionInfo);
+ blockletScanner = new NonFilterScanner(blockExecutionInfo, queryStatisticsModel);
}
if (blockExecutionInfo.isRawRecordDetailQuery()) {
this.scannerResultAggregator =
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/be6bb16d/core/src/main/java/org/apache/carbondata/scan/scanner/AbstractBlockletScanner.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/scan/scanner/AbstractBlockletScanner.java b/core/src/main/java/org/apache/carbondata/scan/scanner/AbstractBlockletScanner.java
index 80e4837..cb6a4ad 100644
--- a/core/src/main/java/org/apache/carbondata/scan/scanner/AbstractBlockletScanner.java
+++ b/core/src/main/java/org/apache/carbondata/scan/scanner/AbstractBlockletScanner.java
@@ -18,6 +18,9 @@
*/
package org.apache.carbondata.scan.scanner;
+import org.apache.carbondata.core.carbon.querystatistics.QueryStatistic;
+import org.apache.carbondata.core.carbon.querystatistics.QueryStatisticsConstants;
+import org.apache.carbondata.core.carbon.querystatistics.QueryStatisticsModel;
import org.apache.carbondata.scan.executor.exception.QueryExecutionException;
import org.apache.carbondata.scan.executor.infos.BlockExecutionInfo;
import org.apache.carbondata.scan.processor.BlocksChunkHolder;
@@ -38,6 +41,8 @@ public abstract class AbstractBlockletScanner implements BlockletScanner {
*/
protected BlockExecutionInfo blockExecutionInfo;
+ public QueryStatisticsModel queryStatisticsModel;
+
public AbstractBlockletScanner(BlockExecutionInfo tableBlockExecutionInfos) {
this.blockExecutionInfo = tableBlockExecutionInfos;
}
@@ -49,6 +54,18 @@ public abstract class AbstractBlockletScanner implements BlockletScanner {
}
protected void fillKeyValue(BlocksChunkHolder blocksChunkHolder) {
+
+ QueryStatistic totalBlockletStatistic = queryStatisticsModel.getStatisticsTypeAndObjMap()
+ .get(QueryStatisticsConstants.TOTAL_BLOCKLET_NUM);
+ totalBlockletStatistic.addCountStatistic(QueryStatisticsConstants.TOTAL_BLOCKLET_NUM,
+ totalBlockletStatistic.getCount() + 1);
+ queryStatisticsModel.getRecorder().recordStatistics(totalBlockletStatistic);
+ QueryStatistic validScannedBlockletStatistic = queryStatisticsModel
+ .getStatisticsTypeAndObjMap().get(QueryStatisticsConstants.VALID_SCAN_BLOCKLET_NUM);
+ validScannedBlockletStatistic
+ .addCountStatistic(QueryStatisticsConstants.VALID_SCAN_BLOCKLET_NUM,
+ validScannedBlockletStatistic.getCount() + 1);
+ queryStatisticsModel.getRecorder().recordStatistics(validScannedBlockletStatistic);
scannedResult.reset();
scannedResult.setNumberOfRows(blocksChunkHolder.getDataBlock().nodeSize());
scannedResult.setDimensionChunks(blocksChunkHolder.getDataBlock()
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/be6bb16d/core/src/main/java/org/apache/carbondata/scan/scanner/impl/NonFilterScanner.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/scan/scanner/impl/NonFilterScanner.java b/core/src/main/java/org/apache/carbondata/scan/scanner/impl/NonFilterScanner.java
index 4d8a8e0..5c5dd3a 100644
--- a/core/src/main/java/org/apache/carbondata/scan/scanner/impl/NonFilterScanner.java
+++ b/core/src/main/java/org/apache/carbondata/scan/scanner/impl/NonFilterScanner.java
@@ -18,6 +18,7 @@
*/
package org.apache.carbondata.scan.scanner.impl;
+import org.apache.carbondata.core.carbon.querystatistics.QueryStatisticsModel;
import org.apache.carbondata.scan.executor.infos.BlockExecutionInfo;
import org.apache.carbondata.scan.result.impl.NonFilterQueryScannedResult;
import org.apache.carbondata.scan.scanner.AbstractBlockletScanner;
@@ -29,9 +30,11 @@ import org.apache.carbondata.scan.scanner.AbstractBlockletScanner;
*/
public class NonFilterScanner extends AbstractBlockletScanner {
- public NonFilterScanner(BlockExecutionInfo blockExecutionInfo) {
+ public NonFilterScanner(BlockExecutionInfo blockExecutionInfo,
+ QueryStatisticsModel queryStatisticsModel) {
super(blockExecutionInfo);
// as its a non filter query creating a non filter query scanned result object
scannedResult = new NonFilterQueryScannedResult(blockExecutionInfo);
+ super.queryStatisticsModel = queryStatisticsModel;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/be6bb16d/core/src/test/java/org/apache/carbondata/core/carbon/querystatistics/QueryStasticsRecorderImplTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/carbondata/core/carbon/querystatistics/QueryStasticsRecorderImplTest.java b/core/src/test/java/org/apache/carbondata/core/carbon/querystatistics/QueryStasticsRecorderImplTest.java
index fd2a55e..90ed930 100644
--- a/core/src/test/java/org/apache/carbondata/core/carbon/querystatistics/QueryStasticsRecorderImplTest.java
+++ b/core/src/test/java/org/apache/carbondata/core/carbon/querystatistics/QueryStasticsRecorderImplTest.java
@@ -39,9 +39,8 @@ public class QueryStasticsRecorderImplTest {
@BeforeClass public static void setUp() {
queryStatisticWithLOAD_BLOCKS_EXECUTOR = new QueryStatistic();
- queryStasticsRecorderImpl = new QueryStatisticsRecorderImpl("query1");
+ queryStasticsRecorderImpl = new QueryStatisticsRecorderImpl(System.nanoTime() + "");
queryStasticsRecorderImpl.logStatisticsAsTableDriver();
- queryStasticsRecorderImpl.logStatisticsAsTableExecutor();
queryStatisticWithLOAD_BLOCKS_EXECUTOR
.addStatistics(QueryStatisticsConstants.LOAD_BLOCKS_EXECUTOR, 5L);
queryStatisticWithLOAD_BLOCKS_EXECUTOR
@@ -101,6 +100,7 @@ public class QueryStasticsRecorderImplTest {
.addStatistics(QueryStatisticsConstants.VALID_SCAN_BLOCKLET_NUM, 5L);
queryStasticsRecorderImpl.logStatistics();
queryStasticsRecorderImpl.recordStatistics(queryStatisticWithVALID_SCAN_BLOCKLET_NUM);
+ queryStasticsRecorderImpl.logStatisticsAsTableExecutor();
}
@Test public void testcollectExecutorStatistics() {
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/be6bb16d/core/src/test/java/org/apache/carbondata/scanner/impl/FilterScannerTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/carbondata/scanner/impl/FilterScannerTest.java b/core/src/test/java/org/apache/carbondata/scanner/impl/FilterScannerTest.java
index debf486..3ff131b 100644
--- a/core/src/test/java/org/apache/carbondata/scanner/impl/FilterScannerTest.java
+++ b/core/src/test/java/org/apache/carbondata/scanner/impl/FilterScannerTest.java
@@ -90,7 +90,7 @@ public class FilterScannerTest {
statisticsTypeAndObjMap.put(QueryStatisticsConstants.TOTAL_BLOCKLET_NUM, queryStatistic);
statisticsTypeAndObjMap.put(QueryStatisticsConstants.VALID_SCAN_BLOCKLET_NUM, queryStatistic);
queryStatisticsModel.setStatisticsTypeAndObjMap(statisticsTypeAndObjMap);
- QueryStatisticsRecorder queryStatisticsRecorder = new QueryStatisticsRecorderImpl("1");
+ QueryStatisticsRecorder queryStatisticsRecorder = new QueryStatisticsRecorderImpl(System.nanoTime() + "");
queryStatisticsModel.setRecorder(queryStatisticsRecorder);
filterScanner = new FilterScanner(blockExecutionInfo, queryStatisticsModel);
blockletIndex = new BlockletIndex();
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/be6bb16d/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanRDD.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanRDD.scala
index 5d972bf..cb3afaa 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanRDD.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanRDD.scala
@@ -36,13 +36,14 @@ import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.carbon.AbsoluteTableIdentifier
import org.apache.carbondata.core.carbon.datastore.block.Distributable
import org.apache.carbondata.core.carbon.metadata.schema.table.CarbonTable
-import org.apache.carbondata.core.carbon.querystatistics.{QueryStatistic, QueryStatisticsConstants}
+import org.apache.carbondata.core.carbon.querystatistics.{QueryStatistic, QueryStatisticsConstants, QueryStatisticsRecorder}
import org.apache.carbondata.core.util.CarbonTimeStatisticsFactory
import org.apache.carbondata.hadoop._
import org.apache.carbondata.scan.expression.Expression
import org.apache.carbondata.scan.model.QueryModel
import org.apache.carbondata.spark.load.CarbonLoaderUtil
+
/**
* This RDD is used to perform query on CarbonData file. Before sending tasks to scan
* CarbonData file, this RDD will leverage CarbonData's index information to do CarbonData file
@@ -147,7 +148,6 @@ class CarbonScanRDD(
noOfBlocks = splits.size
noOfTasks = result.size()
- statistic = new QueryStatistic()
statistic.addStatistics(QueryStatisticsConstants.BLOCK_IDENTIFICATION,
System.currentTimeMillis)
statisticRecorder.recordStatisticsForDriver(statistic, queryId)
@@ -199,10 +199,10 @@ class CarbonScanRDD(
private var finished = false
private var count = 0
- context.addTaskCompletionListener { context =>
- logStatistics(queryStartTime, count)
- reader.close()
- }
+ context.addTaskCompletionListener { context =>
+ logStatistics(queryStartTime, count, model.getStatisticsRecorder)
+ reader.close()
+ }
override def hasNext: Boolean = {
if (context.isInterrupted) {
@@ -258,18 +258,18 @@ class CarbonScanRDD(
format
}
- def logStatistics(queryStartTime: Long, recordCount: Int): Unit = {
+ def logStatistics(queryStartTime: Long, recordCount: Int,
+ recorder: QueryStatisticsRecorder): Unit = {
var queryStatistic = new QueryStatistic()
queryStatistic.addFixedTimeStatistic(QueryStatisticsConstants.EXECUTOR_PART,
System.currentTimeMillis - queryStartTime)
- val statisticRecorder = CarbonTimeStatisticsFactory.createExecutorRecorder(queryId)
- statisticRecorder.recordStatistics(queryStatistic)
+ recorder.recordStatistics(queryStatistic)
// result size
queryStatistic = new QueryStatistic()
queryStatistic.addCountStatistic(QueryStatisticsConstants.RESULT_SIZE, recordCount)
- statisticRecorder.recordStatistics(queryStatistic)
+ recorder.recordStatistics(queryStatistic)
// print executor query statistics for each task_id
- statisticRecorder.logStatisticsAsTableExecutor()
+ recorder.logStatisticsAsTableExecutor()
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/be6bb16d/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonMetastore.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonMetastore.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonMetastore.scala
index f9ad661..e5d4883 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonMetastore.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonMetastore.scala
@@ -117,7 +117,7 @@ class CarbonMetastore(conf: RuntimeConfig, val storePath: String) {
private val nextId = new AtomicLong(0)
def nextQueryId: String = {
- s"query_${nextId.getAndIncrement()}"
+ System.nanoTime() + ""
}
val metadata = loadMetadata(storePath, nextQueryId)
[2/2] incubator-carbondata git commit: [CARBONDATA-573]fix statistics
code and adapt query stastistics to spark2 This closes #461
Posted by ch...@apache.org.
[CARBONDATA-573]fix statistics code and adapt query stastistics to spark2 This closes #461
Project: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/commit/62906181
Tree: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/tree/62906181
Diff: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/diff/62906181
Branch: refs/heads/master
Commit: 629061816463fc3a57ccfbea981f98403bc9398a
Parents: dc7c86e be6bb16
Author: chenliang613 <ch...@apache.org>
Authored: Thu Dec 29 10:04:00 2016 +0800
Committer: chenliang613 <ch...@apache.org>
Committed: Thu Dec 29 10:04:00 2016 +0800
----------------------------------------------------------------------
.../QueryStatisticsRecorderImpl.java | 2 +-
.../processor/AbstractDataBlockIterator.java | 2 +-
.../scan/scanner/AbstractBlockletScanner.java | 17 +++++++++++++++
.../scan/scanner/impl/NonFilterScanner.java | 5 ++++-
.../QueryStasticsRecorderImplTest.java | 4 ++--
.../scanner/impl/FilterScannerTest.java | 2 +-
.../carbondata/spark/rdd/CarbonScanRDD.scala | 22 ++++++++++----------
.../apache/spark/sql/hive/CarbonMetastore.scala | 2 +-
8 files changed, 38 insertions(+), 18 deletions(-)
----------------------------------------------------------------------