You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ra...@apache.org on 2016/09/22 05:36:04 UTC
[06/50] [abbrv] incubator-carbondata git commit: add more query
statistics
add more query statistics
example recovery
show statistics as table
add query id for decoder
add common constants file
print driver log and executor log seperately
add single recorder
and lock for log
fixed example
Project: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/commit/346a869c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/tree/346a869c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/diff/346a869c
Branch: refs/heads/branch-0.1
Commit: 346a869c1b4eac21e78719c976b30fc0f4f45896
Parents: ce34e10
Author: foryou2030 <fo...@126.com>
Authored: Wed Aug 24 19:20:07 2016 +0800
Committer: ravipesala <ra...@gmail.com>
Committed: Thu Sep 22 09:31:05 2016 +0530
----------------------------------------------------------------------
.../DriverQueryStatisticsRecorder.java | 183 +++++++++++++++++++
.../carbon/querystatistics/QueryStatistic.java | 24 +++
.../QueryStatisticsConstants.java | 53 ++++++
.../QueryStatisticsRecorder.java | 90 +++++++++
.../core/util/CarbonTimeStatisticsFactory.java | 13 +-
.../apache/carbondata/core/util/CarbonUtil.java | 15 ++
.../executor/impl/AbstractQueryExecutor.java | 11 +-
.../AbstractDetailQueryResultIterator.java | 30 ++-
.../iterator/DetailQueryResultIterator.java | 2 +
.../carbondata/hadoop/CarbonInputFormat.java | 3 +-
.../carbondata/spark/rdd/CarbonScanRDD.scala | 49 +++--
.../carbondata/spark/util/CarbonScalaUtil.scala | 1 +
.../org/apache/spark/sql/CarbonContext.scala | 15 +-
.../spark/sql/CarbonDictionaryDecoder.scala | 26 ++-
.../org/apache/spark/sql/CarbonOperators.scala | 2 +-
.../spark/sql/hive/CarbonMetastoreCatalog.scala | 12 +-
16 files changed, 491 insertions(+), 38 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/346a869c/core/src/main/java/org/apache/carbondata/core/carbon/querystatistics/DriverQueryStatisticsRecorder.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/carbon/querystatistics/DriverQueryStatisticsRecorder.java b/core/src/main/java/org/apache/carbondata/core/carbon/querystatistics/DriverQueryStatisticsRecorder.java
new file mode 100644
index 0000000..9e83c22
--- /dev/null
+++ b/core/src/main/java/org/apache/carbondata/core/carbon/querystatistics/DriverQueryStatisticsRecorder.java
@@ -0,0 +1,183 @@
+/*
+ * 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.carbon.querystatistics;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.carbondata.common.logging.LogService;
+import org.apache.carbondata.common.logging.LogServiceFactory;
+
+import static org.apache.carbondata.core.util.CarbonUtil.printLine;
+
+/**
+ * Class will be used to record and log the query statistics
+ */
+public class DriverQueryStatisticsRecorder {
+
+ private static final LogService LOGGER =
+ LogServiceFactory.getLogService(DriverQueryStatisticsRecorder.class.getName());
+
+ /**
+ * singleton QueryStatisticsRecorder for driver
+ */
+ private Map<String, List<QueryStatistic>> queryStatisticsMap;
+
+ /**
+ * lock for log statistics table
+ */
+ private static final Object lock = new Object();
+
+ private DriverQueryStatisticsRecorder() {
+ queryStatisticsMap = new HashMap<String, List<QueryStatistic>>();
+ }
+
+ private static DriverQueryStatisticsRecorder carbonLoadStatisticsImplInstance =
+ new DriverQueryStatisticsRecorder();
+
+ public static DriverQueryStatisticsRecorder getInstance() {
+ return carbonLoadStatisticsImplInstance;
+ }
+
+ /**
+ * Below method will be used to add the statistics
+ *
+ * @param statistic
+ */
+ public synchronized void recordStatisticsForDriver(QueryStatistic statistic, String queryId) {
+ // refresh query Statistics Map
+ if (queryStatisticsMap.get(queryId) != null) {
+ queryStatisticsMap.get(queryId).add(statistic);
+ } else {
+ List<QueryStatistic> newQueryStatistics = new ArrayList<QueryStatistic>();
+ newQueryStatistics.add(statistic);
+ queryStatisticsMap.put(queryId, newQueryStatistics);
+ }
+ }
+
+ /**
+ * Below method will be used to show statistic log as table
+ */
+ public void logStatisticsAsTableDriver() {
+ synchronized (lock) {
+ String tableInfo = collectDriverStatistics();
+ if (null != tableInfo) {
+ LOGGER.statistic(tableInfo);
+ }
+ }
+ }
+
+ /**
+ * Below method will parse queryStatisticsMap and put time into table
+ */
+ public String collectDriverStatistics() {
+ for (String key: queryStatisticsMap.keySet()) {
+ try {
+ // TODO: get the finished query, and print Statistics
+ if (queryStatisticsMap.get(key).size() > 3) {
+ String sql_parse_time = "";
+ String load_meta_time = "";
+ String block_allocation_time = "";
+ String block_identification_time = "";
+ Double driver_part_time_tmp = 0.0;
+ String splitChar = " ";
+ // get statistic time from the QueryStatistic
+ for (QueryStatistic statistic : queryStatisticsMap.get(key)) {
+ switch (statistic.getMessage()) {
+ case QueryStatisticsConstants.SQL_PARSE:
+ sql_parse_time += statistic.getTimeTaken() + splitChar;
+ driver_part_time_tmp += statistic.getTimeTaken();
+ break;
+ case QueryStatisticsConstants.LOAD_META:
+ load_meta_time += statistic.getTimeTaken() + splitChar;
+ driver_part_time_tmp += statistic.getTimeTaken();
+ break;
+ case QueryStatisticsConstants.BLOCK_ALLOCATION:
+ block_allocation_time += statistic.getTimeTaken() + splitChar;
+ driver_part_time_tmp += statistic.getTimeTaken();
+ break;
+ case QueryStatisticsConstants.BLOCK_IDENTIFICATION:
+ block_identification_time += statistic.getTimeTaken() + splitChar;
+ driver_part_time_tmp += statistic.getTimeTaken();
+ break;
+ default:
+ break;
+ }
+ }
+ String driver_part_time = driver_part_time_tmp + splitChar;
+ // structure the query statistics info table
+ StringBuilder tableInfo = new StringBuilder();
+ int len1 = 8;
+ int len2 = 20;
+ int len3 = 21;
+ int len4 = 22;
+ String line = "+" + printLine("-", len1) + "+" + printLine("-", len2) + "+" +
+ printLine("-", len3) + "+" + printLine("-", len4) + "+";
+ String line2 = "|" + printLine(" ", len1) + "+" + printLine("-", len2) + "+" +
+ printLine(" ", len3) + "+" + printLine("-", len4) + "+";
+ // table header
+ tableInfo.append(line).append("\n");
+ tableInfo.append("|" + printLine(" ", (len1 - "Module".length())) + "Module" + "|" +
+ printLine(" ", (len2 - "Operation Step".length())) + "Operation Step" + "|" +
+ printLine(" ", (len3 + len4 + 1 - "Query Cost".length())) +
+ "Query Cost" + "|" + "\n");
+ // driver part
+ tableInfo.append(line).append("\n");
+ tableInfo.append("|" + printLine(" ", len1) + "|" +
+ printLine(" ", (len2 - "SQL parse".length())) + "SQL parse" + "|" +
+ printLine(" ", len3) + "|" +
+ printLine(" ", (len4 - sql_parse_time.length())) + sql_parse_time + "|" + "\n");
+ tableInfo.append(line2).append("\n");
+ tableInfo.append("|" +printLine(" ", (len1 - "Driver".length())) + "Driver" + "|" +
+ printLine(" ", (len2 - "Load meta data".length())) + "Load meta data" + "|" +
+ printLine(" ", (len3 - driver_part_time.length())) + driver_part_time + "|" +
+ printLine(" ", (len4 - load_meta_time.length())) +
+ load_meta_time + "|" + "\n");
+ tableInfo.append(line2).append("\n");
+ tableInfo.append("|" +
+ printLine(" ", (len1 - "Part".length())) + "Part" + "|" +
+ printLine(" ", (len2 - "Block allocation".length())) +
+ "Block allocation" + "|" +
+ printLine(" ", len3) + "|" +
+ printLine(" ", (len4 - block_allocation_time.length())) +
+ block_allocation_time + "|" + "\n");
+ tableInfo.append(line2).append("\n");
+ tableInfo.append("|" +
+ printLine(" ", len1) + "|" +
+ printLine(" ", (len2 - "Block identification".length())) +
+ "Block identification" + "|" +
+ printLine(" ", len3) + "|" +
+ printLine(" ", (len4 - block_identification_time.length())) +
+ block_identification_time + "|" + "\n");
+ tableInfo.append(line).append("\n");
+
+ // once the statistics be printed, remove it from the map
+ queryStatisticsMap.remove(key);
+ // show query statistic as "query id" + "table"
+ return "Print query statistic for query id: " + key + "\n" + tableInfo.toString();
+ }
+ } catch (Exception ex) {
+ return "Put statistics into table failed, catch exception: " + ex.getMessage();
+ }
+ }
+ return null;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/346a869c/core/src/main/java/org/apache/carbondata/core/carbon/querystatistics/QueryStatistic.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/carbon/querystatistics/QueryStatistic.java b/core/src/main/java/org/apache/carbondata/core/carbon/querystatistics/QueryStatistic.java
index 39198c8..3e44949 100644
--- a/core/src/main/java/org/apache/carbondata/core/carbon/querystatistics/QueryStatistic.java
+++ b/core/src/main/java/org/apache/carbondata/core/carbon/querystatistics/QueryStatistic.java
@@ -45,6 +45,11 @@ public class QueryStatistic implements Serializable {
*/
private long startTime;
+ /**
+ * number of count
+ */
+ private long count;
+
public QueryStatistic() {
this.startTime = System.currentTimeMillis();
}
@@ -72,6 +77,12 @@ public class QueryStatistic implements Serializable {
this.message = message;
}
+ public void addCountStatistic(String message, long count) {
+ this.timeTaken = -1;
+ this.count = count;
+ this.message = message;
+ }
+
/**
* Below method will be used to get the statistic message, which will
* be used to log
@@ -82,4 +93,17 @@ public class QueryStatistic implements Serializable {
public String getStatistics(String queryWithTaskId) {
return message + " for the taskid : " + queryWithTaskId + " Is : " + timeTaken;
}
+
+ public String getMessage() {
+ return this.message;
+ }
+
+ public double getTimeTaken() {
+ return (double)this.timeTaken/1000;
+ }
+
+ public long getCount() {
+ return this.count;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/346a869c/core/src/main/java/org/apache/carbondata/core/carbon/querystatistics/QueryStatisticsConstants.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/carbon/querystatistics/QueryStatisticsConstants.java b/core/src/main/java/org/apache/carbondata/core/carbon/querystatistics/QueryStatisticsConstants.java
new file mode 100644
index 0000000..540cf1e
--- /dev/null
+++ b/core/src/main/java/org/apache/carbondata/core/carbon/querystatistics/QueryStatisticsConstants.java
@@ -0,0 +1,53 @@
+/*
+ * 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.carbon.querystatistics;
+
+public interface QueryStatisticsConstants {
+
+ // driver side
+ String SQL_PARSE = "Time taken to parse sql In Driver Side";
+
+ String LOAD_META = "Time taken to load meta data In Driver Side";
+
+ String LOAD_BLOCKS_DRIVER = "Time taken to load the Block(s) In Driver Side";
+
+ String BLOCK_ALLOCATION = "Total Time taken in block(s) allocation";
+
+ String BLOCK_IDENTIFICATION = "Time taken to identify Block(s) to scan";
+
+ // executor side
+ String EXECUTOR_PART =
+ "Total Time taken to execute the query in executor Side";
+
+ String LOAD_BLOCKS_EXECUTOR = "Time taken to load the Block(s) In Executor";
+
+ String SCAN_BLOCKS_NUM = "The num of blocks scanned";
+
+ String SCAN_BLOCKS_TIME = "Time taken to scan blocks";
+
+ String LOAD_DICTIONARY = "Time taken to load the Dictionary In Executor";
+
+ String PREPARE_RESULT = "Total Time taken to prepare query result";
+
+ String RESULT_SIZE = "The size of query result";
+
+}
+
+
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/346a869c/core/src/main/java/org/apache/carbondata/core/carbon/querystatistics/QueryStatisticsRecorder.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/carbon/querystatistics/QueryStatisticsRecorder.java b/core/src/main/java/org/apache/carbondata/core/carbon/querystatistics/QueryStatisticsRecorder.java
index ce12cae..e713237 100644
--- a/core/src/main/java/org/apache/carbondata/core/carbon/querystatistics/QueryStatisticsRecorder.java
+++ b/core/src/main/java/org/apache/carbondata/core/carbon/querystatistics/QueryStatisticsRecorder.java
@@ -25,6 +25,8 @@ import java.util.List;
import org.apache.carbondata.common.logging.LogService;
import org.apache.carbondata.common.logging.LogServiceFactory;
+import static org.apache.carbondata.core.util.CarbonUtil.printLine;
+
/**
* Class will be used to record and log the query statistics
*/
@@ -32,6 +34,7 @@ public class QueryStatisticsRecorder implements Serializable {
private static final LogService LOGGER =
LogServiceFactory.getLogService(QueryStatisticsRecorder.class.getName());
+
/**
* serialization version
*/
@@ -49,6 +52,11 @@ public class QueryStatisticsRecorder implements Serializable {
*/
private String queryIWthTask;
+ /**
+ * lock for log statistics table
+ */
+ private static final Object lock = new Object();
+
public QueryStatisticsRecorder(String queryId) {
queryStatistics = new ArrayList<QueryStatistic>();
this.queryIWthTask = queryId;
@@ -71,4 +79,86 @@ public class QueryStatisticsRecorder implements Serializable {
LOGGER.statistic(statistic.getStatistics(queryIWthTask));
}
}
+
+ /**
+ * Below method will be used to show statistic log as table
+ */
+ public void logStatisticsAsTableExecutor() {
+ synchronized (lock) {
+ String tableInfo = collectExecutorStatistics();
+ if (null != tableInfo) {
+ LOGGER.statistic(tableInfo);
+ }
+ }
+ }
+
+ /**
+ * Below method will parse queryStatisticsMap and put time into table
+ */
+ public String collectExecutorStatistics() {
+ String load_blocks_time = "";
+ String scan_blocks_time = "";
+ String scan_blocks_num = "";
+ String load_dictionary_time = "";
+ String result_size = "";
+ String total_executor_time = "";
+ String splitChar = " ";
+ try {
+ for (QueryStatistic statistic : queryStatistics) {
+ switch (statistic.getMessage()) {
+ case QueryStatisticsConstants.LOAD_BLOCKS_EXECUTOR:
+ load_blocks_time += statistic.getTimeTaken() + splitChar;
+ break;
+ case QueryStatisticsConstants.SCAN_BLOCKS_TIME:
+ scan_blocks_time += statistic.getTimeTaken() + splitChar;
+ break;
+ case QueryStatisticsConstants.SCAN_BLOCKS_NUM:
+ scan_blocks_num += statistic.getCount() + splitChar;
+ break;
+ case QueryStatisticsConstants.LOAD_DICTIONARY:
+ load_dictionary_time += statistic.getTimeTaken() + splitChar;
+ break;
+ case QueryStatisticsConstants.RESULT_SIZE:
+ result_size += statistic.getCount() + " ";
+ break;
+ case QueryStatisticsConstants.EXECUTOR_PART:
+ total_executor_time += statistic.getTimeTaken() + splitChar;
+ break;
+ default:
+ break;
+ }
+ }
+ String headers = "task_id,load_blocks_time,load_dictionary_time,scan_blocks_time," +
+ "scan_blocks_num,result_size,total_executor_time";
+ List<String> values = new ArrayList<String>();
+ values.add(queryIWthTask);
+ values.add(load_blocks_time);
+ values.add(load_dictionary_time);
+ values.add(scan_blocks_time);
+ values.add(scan_blocks_num);
+ values.add(result_size);
+ values.add(total_executor_time);
+ StringBuilder tableInfo = new StringBuilder();
+ String[] columns = headers.split(",");
+ String line = "";
+ String hearLine = "";
+ String valueLine = "";
+ for (int i = 0; i < columns.length; i++) {
+ int len = Math.max(columns[i].length(), values.get(i).length());
+ line += "+" + printLine("-", len);
+ hearLine += "|" + printLine(" ", len - columns[i].length()) + columns[i];
+ valueLine += "|" + printLine(" ", len - values.get(i).length()) + values.get(i);
+ }
+ // struct table info
+ tableInfo.append(line + "+").append("\n");
+ tableInfo.append(hearLine + "|").append("\n");
+ tableInfo.append(line + "+").append("\n");
+ tableInfo.append(valueLine + "|").append("\n");
+ tableInfo.append(line + "+").append("\n");
+ return "Print query statistic for each task id:" + "\n" + tableInfo.toString();
+ } catch (Exception ex) {
+ return "Put statistics into table failed, catch exception: " + ex.getMessage();
+ }
+ }
+
}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/346a869c/core/src/main/java/org/apache/carbondata/core/util/CarbonTimeStatisticsFactory.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/util/CarbonTimeStatisticsFactory.java b/core/src/main/java/org/apache/carbondata/core/util/CarbonTimeStatisticsFactory.java
index c7c2b8a..274800f 100644
--- a/core/src/main/java/org/apache/carbondata/core/util/CarbonTimeStatisticsFactory.java
+++ b/core/src/main/java/org/apache/carbondata/core/util/CarbonTimeStatisticsFactory.java
@@ -17,15 +17,18 @@
package org.apache.carbondata.core.util;
+import org.apache.carbondata.core.carbon.querystatistics.DriverQueryStatisticsRecorder;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
public class CarbonTimeStatisticsFactory {
private static String LoadStatisticsInstanceType;
private static LoadStatistics LoadStatisticsInstance;
+ private static DriverQueryStatisticsRecorder QueryStatisticsRecorderInstance;
static {
CarbonTimeStatisticsFactory.updateTimeStatisticsUtilStatus();
LoadStatisticsInstance = genLoadStatisticsInstance();
+ QueryStatisticsRecorderInstance = genQueryStatisticsRecorderInstance();
}
private static void updateTimeStatisticsUtilStatus() {
@@ -49,4 +52,12 @@ public class CarbonTimeStatisticsFactory {
return LoadStatisticsInstance;
}
-}
+ private static DriverQueryStatisticsRecorder genQueryStatisticsRecorderInstance() {
+ return DriverQueryStatisticsRecorder.getInstance();
+ }
+
+ public static DriverQueryStatisticsRecorder getQueryStatisticsRecorderInstance() {
+ return QueryStatisticsRecorderInstance;
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/346a869c/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java b/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java
index df538e0..77d041c 100644
--- a/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java
+++ b/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java
@@ -1415,5 +1415,20 @@ public final class CarbonUtil {
}
return readLine;
}
+
+ /**
+ * Below method will create string like "***********"
+ *
+ * @param a
+ * @param num
+ */
+ public static String printLine(String a, int num)
+ {
+ StringBuilder builder = new StringBuilder();
+ for (int i = 0; i < num; i++) {
+ builder.append(a);
+ }
+ return builder.toString();
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/346a869c/core/src/main/java/org/apache/carbondata/scan/executor/impl/AbstractQueryExecutor.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/scan/executor/impl/AbstractQueryExecutor.java b/core/src/main/java/org/apache/carbondata/scan/executor/impl/AbstractQueryExecutor.java
index 832b2fa..dab8a23 100644
--- a/core/src/main/java/org/apache/carbondata/scan/executor/impl/AbstractQueryExecutor.java
+++ b/core/src/main/java/org/apache/carbondata/scan/executor/impl/AbstractQueryExecutor.java
@@ -38,6 +38,7 @@ import org.apache.carbondata.core.carbon.metadata.encoder.Encoding;
import org.apache.carbondata.core.carbon.metadata.schema.table.column.CarbonDimension;
import org.apache.carbondata.core.carbon.metadata.schema.table.column.CarbonMeasure;
import org.apache.carbondata.core.carbon.querystatistics.QueryStatistic;
+import org.apache.carbondata.core.carbon.querystatistics.QueryStatisticsConstants;
import org.apache.carbondata.core.carbon.querystatistics.QueryStatisticsRecorder;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.datastorage.store.impl.FileFactory;
@@ -108,7 +109,7 @@ public abstract class AbstractQueryExecutor<E> implements QueryExecutor<E> {
throw new QueryExecutionException(e);
}
queryStatistic
- .addStatistics("Time taken to load the Block(s) In Executor", System.currentTimeMillis());
+ .addStatistics(QueryStatisticsConstants.LOAD_BLOCKS_EXECUTOR, System.currentTimeMillis());
queryProperties.queryStatisticsRecorder.recordStatistics(queryStatistic);
//
// // updating the restructuring infos for the query
@@ -150,7 +151,7 @@ public abstract class AbstractQueryExecutor<E> implements QueryExecutor<E> {
.getDimensionDictionaryDetail(queryModel.getQueryDimension(),
queryProperties.complexFilterDimension, queryModel.getAbsoluteTableIdentifier());
queryStatistic
- .addStatistics("Time taken to load the Dictionary In Executor", System.currentTimeMillis());
+ .addStatistics(QueryStatisticsConstants.LOAD_DICTIONARY, System.currentTimeMillis());
queryProperties.queryStatisticsRecorder.recordStatistics(queryStatistic);
queryModel.setColumnToDictionaryMapping(queryProperties.columnToDictionayMapping);
// setting the sort dimension index. as it will be updated while getting the sort info
@@ -207,6 +208,12 @@ public abstract class AbstractQueryExecutor<E> implements QueryExecutor<E> {
}
queryProperties.complexDimensionInfoMap =
blockExecutionInfoList.get(blockExecutionInfoList.size() - 1).getComlexDimensionInfoMap();
+ if (null != queryModel.getStatisticsRecorder()) {
+ QueryStatistic queryStatistic = new QueryStatistic();
+ queryStatistic.addCountStatistic(QueryStatisticsConstants.SCAN_BLOCKS_NUM,
+ blockExecutionInfoList.size());
+ queryModel.getStatisticsRecorder().recordStatistics(queryStatistic);
+ }
return blockExecutionInfoList;
}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/346a869c/core/src/main/java/org/apache/carbondata/scan/result/iterator/AbstractDetailQueryResultIterator.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/scan/result/iterator/AbstractDetailQueryResultIterator.java b/core/src/main/java/org/apache/carbondata/scan/result/iterator/AbstractDetailQueryResultIterator.java
index d583560..36c4ad0 100644
--- a/core/src/main/java/org/apache/carbondata/scan/result/iterator/AbstractDetailQueryResultIterator.java
+++ b/core/src/main/java/org/apache/carbondata/scan/result/iterator/AbstractDetailQueryResultIterator.java
@@ -26,6 +26,9 @@ import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.carbon.datastore.DataRefNode;
import org.apache.carbondata.core.carbon.datastore.DataRefNodeFinder;
import org.apache.carbondata.core.carbon.datastore.impl.btree.BTreeDataRefNodeFinder;
+import org.apache.carbondata.core.carbon.querystatistics.QueryStatistic;
+import org.apache.carbondata.core.carbon.querystatistics.QueryStatisticsConstants;
+import org.apache.carbondata.core.carbon.querystatistics.QueryStatisticsRecorder;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.datastorage.store.FileHolder;
import org.apache.carbondata.core.datastorage.store.impl.FileFactory;
@@ -67,6 +70,21 @@ public abstract class AbstractDetailQueryResultIterator extends CarbonIterator {
protected boolean nextBatch = false;
+ /**
+ * total time scan the blocks
+ */
+ protected long totalScanTime;
+
+ /**
+ * is the statistic recorded
+ */
+ protected boolean isStatisticsRecorded;
+
+ /**
+ * QueryStatisticsRecorder
+ */
+ protected QueryStatisticsRecorder recorder;
+
public AbstractDetailQueryResultIterator(List<BlockExecutionInfo> infos, QueryModel queryModel) {
String batchSizeString =
CarbonProperties.getInstance().getProperty(CarbonCommonConstants.DETAIL_QUERY_BATCH_SIZE);
@@ -80,7 +98,7 @@ public abstract class AbstractDetailQueryResultIterator extends CarbonIterator {
} else {
batchSize = CarbonCommonConstants.DETAIL_QUERY_BATCH_SIZE_DEFAULT;
}
-
+ this.recorder = queryModel.getStatisticsRecorder();
this.blockExecutionInfos = infos;
this.fileReader = FileFactory.getFileHolder(
FileFactory.getFileType(queryModel.getAbsoluteTableIdentifier().getStorePath()));
@@ -111,8 +129,16 @@ public abstract class AbstractDetailQueryResultIterator extends CarbonIterator {
@Override public boolean hasNext() {
if ((dataBlockIterator != null && dataBlockIterator.hasNext()) || nextBatch) {
return true;
+ } else if (blockExecutionInfos.size() > 0) {
+ return true;
} else {
- return blockExecutionInfos.size() > 0;
+ if (!isStatisticsRecorded) {
+ QueryStatistic statistic = new QueryStatistic();
+ statistic.addFixedTimeStatistic(QueryStatisticsConstants.SCAN_BLOCKS_TIME, totalScanTime);
+ recorder.recordStatistics(statistic);
+ isStatisticsRecorded = true;
+ }
+ return false;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/346a869c/core/src/main/java/org/apache/carbondata/scan/result/iterator/DetailQueryResultIterator.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/scan/result/iterator/DetailQueryResultIterator.java b/core/src/main/java/org/apache/carbondata/scan/result/iterator/DetailQueryResultIterator.java
index fa804a5..0013c0a 100644
--- a/core/src/main/java/org/apache/carbondata/scan/result/iterator/DetailQueryResultIterator.java
+++ b/core/src/main/java/org/apache/carbondata/scan/result/iterator/DetailQueryResultIterator.java
@@ -49,6 +49,7 @@ public class DetailQueryResultIterator extends AbstractDetailQueryResultIterator
@Override public BatchResult next() {
BatchResult result;
+ long startTime = System.currentTimeMillis();
try {
if (future == null) {
future = execute();
@@ -63,6 +64,7 @@ public class DetailQueryResultIterator extends AbstractDetailQueryResultIterator
execService.awaitTermination(1, TimeUnit.HOURS);
fileReader.finish();
}
+ totalScanTime += System.currentTimeMillis() - startTime;
} catch (Exception ex) {
execService.shutdown();
fileReader.finish();
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/346a869c/hadoop/src/main/java/org/apache/carbondata/hadoop/CarbonInputFormat.java
----------------------------------------------------------------------
diff --git a/hadoop/src/main/java/org/apache/carbondata/hadoop/CarbonInputFormat.java b/hadoop/src/main/java/org/apache/carbondata/hadoop/CarbonInputFormat.java
index 6bc692f..da697ad 100644
--- a/hadoop/src/main/java/org/apache/carbondata/hadoop/CarbonInputFormat.java
+++ b/hadoop/src/main/java/org/apache/carbondata/hadoop/CarbonInputFormat.java
@@ -49,6 +49,7 @@ import org.apache.carbondata.core.carbon.metadata.schema.table.CarbonTable;
import org.apache.carbondata.core.carbon.path.CarbonStorePath;
import org.apache.carbondata.core.carbon.path.CarbonTablePath;
import org.apache.carbondata.core.carbon.querystatistics.QueryStatistic;
+import org.apache.carbondata.core.carbon.querystatistics.QueryStatisticsConstants;
import org.apache.carbondata.core.carbon.querystatistics.QueryStatisticsRecorder;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.keygenerator.KeyGenException;
@@ -483,7 +484,7 @@ public class CarbonInputFormat<T> extends FileInputFormat<Void, T> {
}
resultFilterredBlocks.addAll(filterredBlocks);
}
- statistic.addStatistics("Time taken to load the Block(s) in Driver Side",
+ statistic.addStatistics(QueryStatisticsConstants.LOAD_BLOCKS_DRIVER,
System.currentTimeMillis());
recorder.recordStatistics(statistic);
recorder.logStatistics();
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/346a869c/integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanRDD.scala b/integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanRDD.scala
index 6c0438f..5f50644 100644
--- a/integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanRDD.scala
+++ b/integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanRDD.scala
@@ -32,7 +32,8 @@ import org.apache.carbondata.common.CarbonIterator
import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.cache.dictionary.Dictionary
import org.apache.carbondata.core.carbon.datastore.block.{BlockletInfos, TableBlockInfo}
-import org.apache.carbondata.core.carbon.querystatistics.{QueryStatistic, QueryStatisticsRecorder}
+import org.apache.carbondata.core.carbon.querystatistics.{QueryStatistic, QueryStatisticsConstants, QueryStatisticsRecorder}
+import org.apache.carbondata.core.util.CarbonTimeStatisticsFactory
import org.apache.carbondata.hadoop.{CarbonInputFormat, CarbonInputSplit}
import org.apache.carbondata.scan.executor.QueryExecutorFactory
import org.apache.carbondata.scan.expression.Expression
@@ -75,8 +76,7 @@ class CarbonScanRDD[V: ClassTag](
val defaultParallelism = sc.defaultParallelism
override def getPartitions: Array[Partition] = {
- val statisticRecorder = new QueryStatisticsRecorder(queryModel.getQueryId)
- val startTime = System.currentTimeMillis()
+ val statisticRecorder = CarbonTimeStatisticsFactory.getQueryStatisticsRecorderInstance()
val (carbonInputFormat: CarbonInputFormat[Array[Object]], job: Job) =
QueryPlanUtil.createCarbonInputFormat(queryModel.getAbsoluteTableIdentifier)
@@ -113,19 +113,17 @@ class CarbonScanRDD[V: ClassTag](
val blockList = CarbonLoaderUtil.
distributeBlockLets(blockListTemp.asJava, defaultParallelism).asScala
if (blockList.nonEmpty) {
+ var statistic = new QueryStatistic()
// group blocks to nodes, tasks
- val startTime = System.currentTimeMillis
- var statistic = new QueryStatistic
val activeNodes = DistributionUtil
.ensureExecutorsAndGetNodeList(blockList.toArray, sparkContext)
val nodeBlockMapping =
CarbonLoaderUtil.nodeBlockTaskMapping(blockList.asJava, -1, defaultParallelism,
activeNodes.toList.asJava
)
- val timeElapsed: Long = System.currentTimeMillis - startTime
- statistic.addStatistics("Total Time taken in block(s) allocation", System.currentTimeMillis)
- statisticRecorder.recordStatistics(statistic);
- statistic = new QueryStatistic
+ statistic.addStatistics(QueryStatisticsConstants.BLOCK_ALLOCATION, System.currentTimeMillis)
+ statisticRecorder.recordStatisticsForDriver(statistic, queryModel.getQueryId())
+ statistic = new QueryStatistic()
var i = 0
// Create Spark Partition for each task and assign blocks
nodeBlockMapping.asScala.foreach { entry =>
@@ -146,9 +144,10 @@ class CarbonScanRDD[V: ClassTag](
+ s"parallelism: $defaultParallelism , " +
s"no.of.nodes: $noOfNodes, no.of.tasks: $noOfTasks"
)
- statistic.addStatistics("Time taken to identify Block(s) to scan", System.currentTimeMillis)
- statisticRecorder.recordStatistics(statistic);
- statisticRecorder.logStatistics
+ statistic.addStatistics(QueryStatisticsConstants.BLOCK_IDENTIFICATION,
+ System.currentTimeMillis)
+ statisticRecorder.recordStatisticsForDriver(statistic, queryModel.getQueryId())
+ statisticRecorder.logStatisticsAsTableDriver()
result.asScala.foreach { r =>
val cp = r.asInstanceOf[CarbonSparkPartition]
logInfo(s"Node : " + cp.locations.toSeq.mkString(",")
@@ -216,13 +215,18 @@ class CarbonScanRDD[V: ClassTag](
if (finished) {
clearDictionaryCache(queryModel.getColumnToDictionaryMapping)
if (null != queryModel.getStatisticsRecorder) {
- val queryStatistic = new QueryStatistic
+ var queryStatistic = new QueryStatistic()
queryStatistic
- .addFixedTimeStatistic("Total Time taken to execute the query in executor Side",
+ .addFixedTimeStatistic(QueryStatisticsConstants.EXECUTOR_PART,
System.currentTimeMillis - queryStartTime
)
- queryModel.getStatisticsRecorder.recordStatistics(queryStatistic);
- queryModel.getStatisticsRecorder.logStatistics();
+ queryModel.getStatisticsRecorder.recordStatistics(queryStatistic)
+ // result size
+ queryStatistic = new QueryStatistic()
+ queryStatistic.addCountStatistic(QueryStatisticsConstants.RESULT_SIZE, recordCount)
+ queryModel.getStatisticsRecorder.recordStatistics(queryStatistic)
+ // print executor query statistics for each task_id
+ queryModel.getStatisticsRecorder.logStatisticsAsTableExecutor()
}
}
!finished
@@ -237,13 +241,18 @@ class CarbonScanRDD[V: ClassTag](
if (queryModel.getLimit != -1 && recordCount >= queryModel.getLimit) {
clearDictionaryCache(queryModel.getColumnToDictionaryMapping)
if (null != queryModel.getStatisticsRecorder) {
- val queryStatistic = new QueryStatistic
+ var queryStatistic = new QueryStatistic()
queryStatistic
- .addFixedTimeStatistic("Total Time taken to execute the query in executor Side",
+ .addFixedTimeStatistic(QueryStatisticsConstants.EXECUTOR_PART,
System.currentTimeMillis - queryStartTime
)
- queryModel.getStatisticsRecorder.recordStatistics(queryStatistic);
- queryModel.getStatisticsRecorder.logStatistics();
+ queryModel.getStatisticsRecorder.recordStatistics(queryStatistic)
+ // result size
+ queryStatistic = new QueryStatistic()
+ queryStatistic.addCountStatistic(QueryStatisticsConstants.RESULT_SIZE, recordCount)
+ queryModel.getStatisticsRecorder.recordStatistics(queryStatistic)
+ // print executor query statistics for each task_id
+ queryModel.getStatisticsRecorder.logStatisticsAsTableExecutor()
}
}
keyClass.getValue(rowIterator.next())
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/346a869c/integration/spark/src/main/scala/org/apache/carbondata/spark/util/CarbonScalaUtil.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/carbondata/spark/util/CarbonScalaUtil.scala b/integration/spark/src/main/scala/org/apache/carbondata/spark/util/CarbonScalaUtil.scala
index bdb2a69..028c52f 100644
--- a/integration/spark/src/main/scala/org/apache/carbondata/spark/util/CarbonScalaUtil.scala
+++ b/integration/spark/src/main/scala/org/apache/carbondata/spark/util/CarbonScalaUtil.scala
@@ -163,4 +163,5 @@ object CarbonScalaUtil extends Logging {
}
kettleHomePath
}
+
}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/346a869c/integration/spark/src/main/scala/org/apache/spark/sql/CarbonContext.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/CarbonContext.scala b/integration/spark/src/main/scala/org/apache/spark/sql/CarbonContext.scala
index 3cfe301..972de05 100644
--- a/integration/spark/src/main/scala/org/apache/spark/sql/CarbonContext.scala
+++ b/integration/spark/src/main/scala/org/apache/spark/sql/CarbonContext.scala
@@ -34,8 +34,9 @@ import org.apache.spark.sql.hive._
import org.apache.spark.sql.optimizer.CarbonOptimizer
import org.apache.carbondata.common.logging.LogServiceFactory
+import org.apache.carbondata.core.carbon.querystatistics.{QueryStatistic, QueryStatisticsConstants, QueryStatisticsRecorder}
import org.apache.carbondata.core.constants.CarbonCommonConstants
-import org.apache.carbondata.core.util.CarbonProperties
+import org.apache.carbondata.core.util.{CarbonProperties, CarbonTimeStatisticsFactory}
import org.apache.carbondata.spark.rdd.CarbonDataFrameRDD
class CarbonContext(
@@ -67,12 +68,13 @@ class CarbonContext(
override lazy val catalog = {
CarbonProperties.getInstance()
.addProperty(CarbonCommonConstants.STORE_LOCATION, storePath)
- new CarbonMetastoreCatalog(this, storePath, metadataHive) with OverrideCatalog
+ new CarbonMetastoreCatalog(this, storePath, metadataHive, queryId) with OverrideCatalog
}
@transient
override protected[sql] lazy val analyzer =
new Analyzer(catalog, functionRegistry, conf) {
+
override val extendedResolutionRules =
catalog.ParquetConversions ::
catalog.CreateTables ::
@@ -118,15 +120,21 @@ class CarbonContext(
@transient
val LOGGER = LogServiceFactory.getLogService(CarbonContext.getClass.getName)
+ var queryId: String = ""
+
override def sql(sql: String): DataFrame = {
// queryId will be unique for each query, creting query detail holder
- val queryId: String = System.nanoTime() + ""
+ queryId = System.nanoTime() + ""
this.setConf("queryId", queryId)
CarbonContext.updateCarbonPorpertiesPath(this)
val sqlString = sql.toUpperCase
LOGGER.info(s"Query [$sqlString]")
+ val recorder = CarbonTimeStatisticsFactory.getQueryStatisticsRecorderInstance()
+ val statistic = new QueryStatistic()
val logicPlan: LogicalPlan = parseSql(sql)
+ statistic.addStatistics(QueryStatisticsConstants.SQL_PARSE, System.currentTimeMillis())
+ recorder.recordStatisticsForDriver(statistic, queryId)
val result = new CarbonDataFrameRDD(this, logicPlan)
// We force query optimization to happen right away instead of letting it happen lazily like
@@ -218,6 +226,7 @@ object CarbonContext {
/**
*
* Requesting the extra executors other than the existing ones.
+ *
* @param sc
* @param numExecutors
* @return
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/346a869c/integration/spark/src/main/scala/org/apache/spark/sql/CarbonDictionaryDecoder.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/CarbonDictionaryDecoder.scala b/integration/spark/src/main/scala/org/apache/spark/sql/CarbonDictionaryDecoder.scala
index e4fb857..b29b2e8 100644
--- a/integration/spark/src/main/scala/org/apache/spark/sql/CarbonDictionaryDecoder.scala
+++ b/integration/spark/src/main/scala/org/apache/spark/sql/CarbonDictionaryDecoder.scala
@@ -32,6 +32,7 @@ import org.apache.carbondata.core.carbon.{AbsoluteTableIdentifier, ColumnIdentif
import org.apache.carbondata.core.carbon.metadata.datatype.DataType
import org.apache.carbondata.core.carbon.metadata.encoder.Encoding
import org.apache.carbondata.core.carbon.metadata.schema.table.column.CarbonDimension
+import org.apache.carbondata.core.carbon.querystatistics.{QueryStatistic, QueryStatisticsConstants, QueryStatisticsRecorder}
import org.apache.carbondata.core.util.DataTypeUtil
/**
@@ -147,14 +148,17 @@ case class CarbonDictionaryDecoder(
override def canProcessSafeRows: Boolean = true
+
+
override def doExecute(): RDD[InternalRow] = {
attachTree(this, "execute") {
val storePath = sqlContext.catalog.asInstanceOf[CarbonMetastoreCatalog].storePath
+ val queryId = sqlContext.getConf("queryId", System.nanoTime() + "")
val absoluteTableIdentifiers = relations.map { relation =>
val carbonTable = relation.carbonRelation.carbonRelation.metaData.carbonTable
(carbonTable.getFactTableName, carbonTable.getAbsoluteTableIdentifier)
}.toMap
-
+ val recorder = new QueryStatisticsRecorder(queryId)
if (isRequiredToDecode) {
val dataTypes = child.output.map { attr => attr.dataType }
child.execute().mapPartitions { iter =>
@@ -166,9 +170,21 @@ case class CarbonDictionaryDecoder(
val dictIndex = dicts.zipWithIndex.filter(x => x._1 != null).map(x => x._2)
new Iterator[InternalRow] {
val unsafeProjection = UnsafeProjection.create(output.map(_.dataType).toArray)
- override final def hasNext: Boolean = iter.hasNext
-
+ var flag = true
+ var total = 0L
+ override final def hasNext: Boolean = {
+ flag = iter.hasNext
+ if (false == flag && total > 0) {
+ val queryStatistic = new QueryStatistic()
+ queryStatistic
+ .addFixedTimeStatistic(QueryStatisticsConstants.PREPARE_RESULT, total)
+ recorder.recordStatistics(queryStatistic)
+ recorder.logStatistics()
+ }
+ flag
+ }
override final def next(): InternalRow = {
+ val startTime = System.currentTimeMillis()
val row: InternalRow = iter.next()
val data = row.toSeq(dataTypes).toArray
dictIndex.foreach { index =>
@@ -178,7 +194,9 @@ case class CarbonDictionaryDecoder(
getDictionaryColumnIds(index)._3)
}
}
- unsafeProjection(new GenericMutableRow(data))
+ val result = unsafeProjection(new GenericMutableRow(data))
+ total += System.currentTimeMillis() - startTime
+ result
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/346a869c/integration/spark/src/main/scala/org/apache/spark/sql/CarbonOperators.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/CarbonOperators.scala b/integration/spark/src/main/scala/org/apache/spark/sql/CarbonOperators.scala
index 1452aa2..c2e85d1 100644
--- a/integration/spark/src/main/scala/org/apache/spark/sql/CarbonOperators.scala
+++ b/integration/spark/src/main/scala/org/apache/spark/sql/CarbonOperators.scala
@@ -55,7 +55,7 @@ case class CarbonScan(
plan.setOutLocationPath(
CarbonProperties.getInstance().getProperty(CarbonCommonConstants.STORE_LOCATION_HDFS))
- plan.setQueryId(System.nanoTime() + "")
+ plan.setQueryId(ocRaw.getConf("queryId", System.nanoTime() + ""))
processFilterExpressions(plan)
plan
}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/346a869c/integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonMetastoreCatalog.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonMetastoreCatalog.scala b/integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonMetastoreCatalog.scala
index a9d6077..8db9800 100644
--- a/integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonMetastoreCatalog.scala
+++ b/integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonMetastoreCatalog.scala
@@ -40,12 +40,13 @@ import org.apache.carbondata.core.carbon.metadata.CarbonMetadata
import org.apache.carbondata.core.carbon.metadata.converter.ThriftWrapperSchemaConverterImpl
import org.apache.carbondata.core.carbon.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.carbon.path.{CarbonStorePath, CarbonTablePath}
+import org.apache.carbondata.core.carbon.querystatistics.{QueryStatistic, QueryStatisticsConstants, QueryStatisticsRecorder}
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.datastorage.store.filesystem.CarbonFile
import org.apache.carbondata.core.datastorage.store.impl.FileFactory
import org.apache.carbondata.core.datastorage.store.impl.FileFactory.FileType
import org.apache.carbondata.core.reader.ThriftReader
-import org.apache.carbondata.core.util.{CarbonProperties, CarbonUtil}
+import org.apache.carbondata.core.util.{CarbonProperties, CarbonTimeStatisticsFactory, CarbonUtil}
import org.apache.carbondata.core.writer.ThriftWriter
import org.apache.carbondata.format.{SchemaEvolutionEntry, TableInfo}
import org.apache.carbondata.lcm.locks.ZookeeperInit
@@ -99,7 +100,7 @@ case class DictionaryMap(dictionaryMap: Map[String, Boolean]) {
}
class CarbonMetastoreCatalog(hiveContext: HiveContext, val storePath: String,
- client: ClientInterface)
+ client: ClientInterface, queryId: String)
extends HiveMetastoreCatalog(client, hiveContext)
with spark.Logging {
@@ -151,7 +152,8 @@ class CarbonMetastoreCatalog(hiveContext: HiveContext, val storePath: String,
}
def loadMetadata(metadataPath: String): MetaData = {
-
+ val recorder = CarbonTimeStatisticsFactory.getQueryStatisticsRecorderInstance()
+ val statistic = new QueryStatistic()
// creating zookeeper instance once.
// if zookeeper is configured as carbon lock type.
val zookeeperUrl: String = hiveContext.getConf(CarbonCommonConstants.ZOOKEEPER_URL, null)
@@ -176,8 +178,10 @@ class CarbonMetastoreCatalog(hiveContext: HiveContext, val storePath: String,
val metaDataBuffer = new ArrayBuffer[TableMeta]
fillMetaData(metadataPath, fileType, metaDataBuffer)
updateSchemasUpdatedTime("", "")
+ statistic.addStatistics(QueryStatisticsConstants.LOAD_META,
+ System.currentTimeMillis())
+ recorder.recordStatisticsForDriver(statistic, queryId)
MetaData(metaDataBuffer)
-
}
private def fillMetaData(basePath: String, fileType: FileType,