core/src/main/java/org/apache/carbondata/core/stats/DriverQueryStatisticsRecorderImpl.java - carbondata - Git at Google

 /*
  * 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.stats;

 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;

 import org.apache.carbondata.common.logging.LogServiceFactory;
 import org.apache.carbondata.common.logging.impl.StatisticLevel;

 import static org.apache.carbondata.core.util.CarbonUtil.printLine;

 import org.apache.commons.lang3.StringUtils;
 import org.apache.log4j.Logger;

 /**
  * Class will be used to record and log the query statistics
  */
 public class DriverQueryStatisticsRecorderImpl implements QueryStatisticsRecorder {

   private static final Logger LOGGER =
       LogServiceFactory.getLogService(DriverQueryStatisticsRecorderImpl.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 DriverQueryStatisticsRecorderImpl() {
     // use ConcurrentHashMap, it is thread-safe
     queryStatisticsMap = new ConcurrentHashMap<String, List<QueryStatistic>>();
   }

   private static DriverQueryStatisticsRecorderImpl carbonLoadStatisticsImplInstance =
       new DriverQueryStatisticsRecorderImpl();

   public static DriverQueryStatisticsRecorderImpl getInstance() {
     return carbonLoadStatisticsImplInstance;
   }

   public void recordStatistics(QueryStatistic statistic) {

   }

   public void logStatistics() {

   }

   public TaskStatistics statisticsForTask(long taskId, long startTime) {
     return null;
   }

   public void logStatisticsForTask(TaskStatistics task) {

   }

   /**
    * Below method will be used to add the statistics
    *
    * @param statistic
    */
   public void recordStatisticsForDriver(QueryStatistic statistic, String queryId) {
     if (null != queryId) {
       synchronized (lock) {
         // 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) {
       Iterator<Map.Entry<String, List<QueryStatistic>>> entries =
               queryStatisticsMap.entrySet().iterator();
       while (entries.hasNext()) {
         Map.Entry<String, List<QueryStatistic>> entry = entries.next();
         String queryId = entry.getKey();
         // clear the unknown query statistics
         if (StringUtils.isEmpty(queryId)) {
           entries.remove();
         } else {
           // clear the timeout query statistics
           long interval = System.nanoTime() - Long.parseLong(queryId);
           if (interval > QueryStatisticsConstants.CLEAR_STATISTICS_TIMEOUT) {
             entries.remove();
           } else {
             // print sql_parse_t,load_meta_t,block_allocation_t,block_identification_t
             // or just print block_allocation_t,block_identification_t
             if (entry.getValue().size() >= 2) {
               String tableInfo = collectDriverStatistics(entry.getValue(), queryId);
               if (null != tableInfo) {
                 LOGGER.log(StatisticLevel.STATISTIC, tableInfo);
                 // clear the statistics that has been printed
                 entries.remove();
               }
             }
           }
         }
       }
     }
   }

   /**
    * Below method will parse queryStatisticsMap and put time into table
    */
   public String collectDriverStatistics(List<QueryStatistic> statisticsList, String queryId) {
     String sql_parse_time = "";
     String load_meta_time = "";
     String load_blocks_time = "";
     String block_allocation_time = "";
     String block_identification_time = "";
     long driver_part_time_tmp = 0L;
     long driver_part_time_tmp2 = 0L;
     long load_blocks_time_tmp = 0L;
     String splitChar = " ";
     try {
       // get statistic time from the QueryStatistic
       for (QueryStatistic statistic : statisticsList) {
         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.LOAD_BLOCKS_DRIVER:
             // multi segments will generate multi load_blocks_time
             load_blocks_time_tmp += statistic.getTimeTaken();
             driver_part_time_tmp += statistic.getTimeTaken();
             driver_part_time_tmp2 += statistic.getTimeTaken();
             break;
           case QueryStatisticsConstants.BLOCK_ALLOCATION:
             block_allocation_time += statistic.getTimeTaken() + splitChar;
             driver_part_time_tmp += statistic.getTimeTaken();
             driver_part_time_tmp2 += statistic.getTimeTaken();
             break;
           case QueryStatisticsConstants.BLOCK_IDENTIFICATION:
             block_identification_time += statistic.getTimeTaken() + splitChar;
             driver_part_time_tmp += statistic.getTimeTaken();
             driver_part_time_tmp2 += statistic.getTimeTaken();
             break;
           default:
             break;
         }
       }
       load_blocks_time = load_blocks_time_tmp + splitChar;
       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 = 24;
       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 - "Total Query Cost".length())) + "Total Query Cost" + "|" +
           printLine(" ", (len4 - "Query Cost".length())) + "Query Cost" + "|" + "\n");
       tableInfo.append(line).append("\n");
       // print sql_parse_t,load_meta_t,block_allocation_t,block_identification_t
       if (!StringUtils.isEmpty(sql_parse_time) &&
           !StringUtils.isEmpty(load_meta_time) &&
           !StringUtils.isEmpty(block_allocation_time) &&
           !StringUtils.isEmpty(block_identification_time)) {
         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 - "Load blocks driver".length())) +
                 "Load blocks driver" + "|" +
                 printLine(" ", len3) + "|" +
                 printLine(" ", (len4 - load_blocks_time.length())) +
                 load_blocks_time + "|" + "\n");
         tableInfo.append(line2).append("\n");
         tableInfo.append("|" + printLine(" ", len1) + "|" +
             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");

         // show query statistic as "query id" + "table"
         return "Print query statistic for query id: " + queryId + "\n" + tableInfo.toString();
       } else if (!StringUtils.isEmpty(block_allocation_time) &&
           !StringUtils.isEmpty(block_identification_time)) {
         // when we can't get sql parse time, we only print the last two
         driver_part_time = driver_part_time_tmp2 + splitChar;
         tableInfo.append("|" + printLine(" ", (len1 - "Driver".length())) + "Driver" + "|" +
                 printLine(" ", (len2 - "Load blocks driver".length())) +
                 "Load blocks driver" + "|" +
                 printLine(" ", len3) + "|" +
                 printLine(" ", (len4 - load_blocks_time.length())) +
                 load_blocks_time + "|" + "\n");
         tableInfo.append(line2).append("\n");
         tableInfo.append("|" + printLine(" ", (len1 - "Part".length())) + "Part" + "|" +
             printLine(" ", (len2 - "Block allocation".length())) + "Block allocation" + "|" +
             printLine(" ", (len3 - driver_part_time.length())) + driver_part_time + "|" +
             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");

         // show query statistic as "query id" + "table"
         return "Print query statistic for query id: " + queryId + "\n" + tableInfo.toString();
       }

       return null;
     } catch (Exception ex) {
       return "Put statistics into table failed, catch exception: " + ex.getMessage();
     }
   }
 }
	/*
	* 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.stats;

	import java.util.ArrayList;
	import java.util.Iterator;
	import java.util.List;
	import java.util.Map;
	import java.util.concurrent.ConcurrentHashMap;

	import org.apache.carbondata.common.logging.LogServiceFactory;
	import org.apache.carbondata.common.logging.impl.StatisticLevel;

	import static org.apache.carbondata.core.util.CarbonUtil.printLine;

	import org.apache.commons.lang3.StringUtils;
	import org.apache.log4j.Logger;

	/**
	* Class will be used to record and log the query statistics
	*/
	public class DriverQueryStatisticsRecorderImpl implements QueryStatisticsRecorder {

	private static final Logger LOGGER =
	LogServiceFactory.getLogService(DriverQueryStatisticsRecorderImpl.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 DriverQueryStatisticsRecorderImpl() {
	// use ConcurrentHashMap, it is thread-safe
	queryStatisticsMap = new ConcurrentHashMap<String, List<QueryStatistic>>();
	}

	private static DriverQueryStatisticsRecorderImpl carbonLoadStatisticsImplInstance =
	new DriverQueryStatisticsRecorderImpl();

	public static DriverQueryStatisticsRecorderImpl getInstance() {
	return carbonLoadStatisticsImplInstance;
	}

	public void recordStatistics(QueryStatistic statistic) {

	}

	public void logStatistics() {

	}

	public TaskStatistics statisticsForTask(long taskId, long startTime) {
	return null;
	}

	public void logStatisticsForTask(TaskStatistics task) {

	}

	/**
	* Below method will be used to add the statistics
	*
	* @param statistic
	*/
	public void recordStatisticsForDriver(QueryStatistic statistic, String queryId) {
	if (null != queryId) {
	synchronized (lock) {
	// 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) {
	Iterator<Map.Entry<String, List<QueryStatistic>>> entries =
	queryStatisticsMap.entrySet().iterator();
	while (entries.hasNext()) {
	Map.Entry<String, List<QueryStatistic>> entry = entries.next();
	String queryId = entry.getKey();
	// clear the unknown query statistics
	if (StringUtils.isEmpty(queryId)) {
	entries.remove();
	} else {
	// clear the timeout query statistics
	long interval = System.nanoTime() - Long.parseLong(queryId);
	if (interval > QueryStatisticsConstants.CLEAR_STATISTICS_TIMEOUT) {
	entries.remove();
	} else {
	// print sql_parse_t,load_meta_t,block_allocation_t,block_identification_t
	// or just print block_allocation_t,block_identification_t
	if (entry.getValue().size() >= 2) {
	String tableInfo = collectDriverStatistics(entry.getValue(), queryId);
	if (null != tableInfo) {
	LOGGER.log(StatisticLevel.STATISTIC, tableInfo);
	// clear the statistics that has been printed
	entries.remove();
	}
	}
	}
	}
	}
	}
	}

	/**
	* Below method will parse queryStatisticsMap and put time into table
	*/
	public String collectDriverStatistics(List<QueryStatistic> statisticsList, String queryId) {
	String sql_parse_time = "";
	String load_meta_time = "";
	String load_blocks_time = "";
	String block_allocation_time = "";
	String block_identification_time = "";
	long driver_part_time_tmp = 0L;
	long driver_part_time_tmp2 = 0L;
	long load_blocks_time_tmp = 0L;
	String splitChar = " ";
	try {
	// get statistic time from the QueryStatistic
	for (QueryStatistic statistic : statisticsList) {
	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.LOAD_BLOCKS_DRIVER:
	// multi segments will generate multi load_blocks_time
	load_blocks_time_tmp += statistic.getTimeTaken();
	driver_part_time_tmp += statistic.getTimeTaken();
	driver_part_time_tmp2 += statistic.getTimeTaken();
	break;
	case QueryStatisticsConstants.BLOCK_ALLOCATION:
	block_allocation_time += statistic.getTimeTaken() + splitChar;
	driver_part_time_tmp += statistic.getTimeTaken();
	driver_part_time_tmp2 += statistic.getTimeTaken();
	break;
	case QueryStatisticsConstants.BLOCK_IDENTIFICATION:
	block_identification_time += statistic.getTimeTaken() + splitChar;
	driver_part_time_tmp += statistic.getTimeTaken();
	driver_part_time_tmp2 += statistic.getTimeTaken();
	break;
	default:
	break;
	}
	}
	load_blocks_time = load_blocks_time_tmp + splitChar;
	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 = 24;
	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 - "Total Query Cost".length())) + "Total Query Cost" + "\|" +
	printLine(" ", (len4 - "Query Cost".length())) + "Query Cost" + "\|" + "\n");
	tableInfo.append(line).append("\n");
	// print sql_parse_t,load_meta_t,block_allocation_t,block_identification_t
	if (!StringUtils.isEmpty(sql_parse_time) &&
	!StringUtils.isEmpty(load_meta_time) &&
	!StringUtils.isEmpty(block_allocation_time) &&
	!StringUtils.isEmpty(block_identification_time)) {
	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 - "Load blocks driver".length())) +
	"Load blocks driver" + "\|" +
	printLine(" ", len3) + "\|" +
	printLine(" ", (len4 - load_blocks_time.length())) +
	load_blocks_time + "\|" + "\n");
	tableInfo.append(line2).append("\n");
	tableInfo.append("\|" + printLine(" ", len1) + "\|" +
	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");

	// show query statistic as "query id" + "table"
	return "Print query statistic for query id: " + queryId + "\n" + tableInfo.toString();
	} else if (!StringUtils.isEmpty(block_allocation_time) &&
	!StringUtils.isEmpty(block_identification_time)) {
	// when we can't get sql parse time, we only print the last two
	driver_part_time = driver_part_time_tmp2 + splitChar;
	tableInfo.append("\|" + printLine(" ", (len1 - "Driver".length())) + "Driver" + "\|" +
	printLine(" ", (len2 - "Load blocks driver".length())) +
	"Load blocks driver" + "\|" +
	printLine(" ", len3) + "\|" +
	printLine(" ", (len4 - load_blocks_time.length())) +
	load_blocks_time + "\|" + "\n");
	tableInfo.append(line2).append("\n");
	tableInfo.append("\|" + printLine(" ", (len1 - "Part".length())) + "Part" + "\|" +
	printLine(" ", (len2 - "Block allocation".length())) + "Block allocation" + "\|" +
	printLine(" ", (len3 - driver_part_time.length())) + driver_part_time + "\|" +
	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");

	// show query statistic as "query id" + "table"
	return "Print query statistic for query id: " + queryId + "\n" + tableInfo.toString();
	}

	return null;
	} catch (Exception ex) {
	return "Put statistics into table failed, catch exception: " + ex.getMessage();
	}
	}
	}