/*
 *  Copyright (c) 2017, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 *  WSO2 Inc. 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.wso2.carbon.status.dashboard.core.dbhandler;

import com.zaxxer.hikari.HikariDataSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.wso2.carbon.config.ConfigurationException;
import org.wso2.carbon.database.query.manager.exception.QueryMappingNotAvailableException;
import org.wso2.carbon.status.dashboard.core.bean.table.Attribute;
import org.wso2.carbon.status.dashboard.core.bean.table.ComponentMetrics;
import org.wso2.carbon.status.dashboard.core.bean.table.MetricElement;
import org.wso2.carbon.status.dashboard.core.bean.table.TypeMetrics;
import org.wso2.carbon.status.dashboard.core.dbhandler.utils.DBTableUtils;
import org.wso2.carbon.status.dashboard.core.dbhandler.utils.SQLConstants;
import org.wso2.carbon.status.dashboard.core.exception.RDBMSTableException;
import org.wso2.carbon.status.dashboard.core.exception.StatusDashboardRuntimeException;
import org.wso2.carbon.status.dashboard.core.internal.DashboardDataHolder;

import java.io.IOException;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;

import static org.wso2.carbon.status.dashboard.core.dbhandler.utils.SQLConstants.PERCENTAGE_MARK;
import static org.wso2.carbon.status.dashboard.core.dbhandler.utils.SQLConstants.PLACEHOLDER_AGGREGATION_COMPONENT_COLOUM;
import static org.wso2.carbon.status.dashboard.core.dbhandler.utils.SQLConstants.PLACEHOLDER_AGGREGATION_TIME;
import static org.wso2.carbon.status.dashboard.core.dbhandler.utils.SQLConstants.PLACEHOLDER_COLUMNS;
import static org.wso2.carbon.status.dashboard.core.dbhandler.utils.SQLConstants.PLACEHOLDER_NAME;
import static org.wso2.carbon.status.dashboard.core.dbhandler.utils.SQLConstants.PLACEHOLDER_RESULT;
import static org.wso2.carbon.status.dashboard.core.dbhandler.utils.SQLConstants.PLACEHOLDER_TABLE_NAME;
import static org.wso2.carbon.status.dashboard.core.dbhandler.utils.SQLConstants.PLACEHOLDER_WORKER_ID;
import static org.wso2.carbon.status.dashboard.core.dbhandler.utils.SQLConstants.QUESTION_MARK;

/**
 * This class represents key database operations related to metrics data.
 */
public class StatusDashboardMetricsDBHandler {
    private static final Logger logger = LoggerFactory.getLogger(StatusDashboardMetricsDBHandler.class);
    private static final String DATASOURCE_ID = DashboardDataHolder.getInstance().getStatusDashboardDeploymentConfigs()
            .getMetricsDatasourceName();
    private static final String[] METRICS_TABLE_NAMES = {"METRIC_COUNTER", "METRIC_GAUGE", "METRIC_HISTOGRAM",
            "METRIC_METER", "METRIC_TIMER"};
    private static final String APP_NAME_PREFIX = "org.wso2.siddhi.SiddhiApps.";
    private String selectAppMetricsQuery;
    private String recordSelectAgregatedAppMetricsQuery;
    private String selectWorkerMetricsQuery;
    private String selectWorkerAggregatedMetricsQuery;
    private String selectAppComponentList;
    private String selectAppComponentMetrics;
    private String selectWorkerAggregatedThroughputQuery;
    private String selectWorkerThroughputQuery;
    private String selectAppComponentHistory;
    private String selectAppComponentAggregatedHistory;
    private HikariDataSource dataSource = null;
    private Connection conn;
    private Map<String, Map<String, String>> workerAttributeTypeMap;
    private  QueryManager metricsQueryManager;
    public StatusDashboardMetricsDBHandler() {
        dataSource = DashboardDataHolder.getInstance().getMetricsDataSource();
        if (dataSource != null) {
            try {
                conn= DashboardDataHolder.getInstance().getMetricsDataSource().getConnection();
                DatabaseMetaData databaseMetaData = conn.getMetaData();
                metricsQueryManager = new QueryManager(databaseMetaData.getDatabaseProductName(),
                        databaseMetaData.getDatabaseProductVersion());
                workerAttributeTypeMap = DBTableUtils.getInstance().loadMetricsAttributeTypeMap(metricsQueryManager);
                selectAppMetricsQuery = metricsQueryManager.getQuery(SQLConstants.SELECT_APP_METRICS_QUERY);
                selectWorkerMetricsQuery = metricsQueryManager.getQuery(SQLConstants.SELECT_WORKER_METRICS_QUERY);
                selectWorkerThroughputQuery = metricsQueryManager.getQuery(SQLConstants.
                        SELECT_WORKER_THROUGHPUT_QUERY);
                selectWorkerAggregatedMetricsQuery = metricsQueryManager.getQuery(SQLConstants
                        .SELECT_WORKER_AGGREGATE_METRICS_QUERY);
                selectWorkerAggregatedThroughputQuery = metricsQueryManager.getQuery(SQLConstants.
                        SELECT_WORKER_AGGREGATE_THROUGHPUT_QUERY);
                selectAppComponentList = metricsQueryManager.getQuery(SQLConstants.
                        SELECT_COMPONENT_LIST);
                selectAppComponentMetrics = metricsQueryManager.getQuery(SQLConstants.
                        SELECT_COMPONENT_METRICS);
                selectAppComponentHistory = metricsQueryManager.getQuery(SQLConstants.
                        SELECT_COMPONENT_METRICS_HISTORY);
                recordSelectAgregatedAppMetricsQuery = metricsQueryManager.getQuery(SQLConstants.
                        SELECT_APP_AGG_METRICS_HISTORY);
                selectAppComponentAggregatedHistory =  metricsQueryManager.getQuery(SQLConstants.
                        SELECT_COMPONENT_AGG_METRICS_HISTORY);
            } catch (SQLException | ConfigurationException | IOException | QueryMappingNotAvailableException e) {
                throw new StatusDashboardRuntimeException("Error initializing connection. ", e);
            } finally {
                if (conn != null) {
                    try {
                        conn.close();
                    } catch (SQLException e) {
                        logger.warn("Database error. Could not close database connection", e);
                    }
                }
            }
        } else {
            logger.warn(DATASOURCE_ID + " Could not find. Hence cannot initialize the status dashboard.");
        }
    }


    /**
     * This resolve the table name in generic tables.
     *
     * @param query loaded queries.
     * @return table name resolving.
     */
    private String resolveTableName(String query, String tableName) {
        return query.replace(PLACEHOLDER_TABLE_NAME, tableName);
    }

    /**
     * Returns a connection instance.
     *
     * @return a new {@link Connection} instance from the datasource.
     */
    private Connection getConnection() {
        try {
            if ((conn != null) && (!conn.isClosed())) {
            } else {
                try {
                    this.conn = this.dataSource.getConnection();
                    this.conn.setAutoCommit(true);
                } catch (SQLException e) {
                    throw new RDBMSTableException("Error reinitializing connection: " + e.getMessage() + " in "
                            + DATASOURCE_ID, e);
                }
            }
        } catch (SQLException e) {
            throw new RDBMSTableException("Error while getting connection ", e);
        }
        return conn;
    }

    /**
     * Select the component History .
     *
     * @param workerId         ID of the worker
     * @param appName          siddhi application name
     * @param timeInterval     time interval that needed to be taken.
     * @param currentTimeMilli current time in milliseconds expression in db type
     * @param metricsType      table name to be fetched
     * @return
     */
    public List<List<Object>> selectAppComponentsHistory(String workerId, String appName, long timeInterval, long
            currentTimeMilli, String metricsType, String componentType, String componentId) {
        Map<String, String> typeTableColumn = DBTableUtils.getInstance().loadMetricsTypeSelection();
        String tableName = typeTableColumn.get(metricsType);
        Map<String, String> tableColumn = DBTableUtils.getInstance().loadMetricsAllValueSelection();
        String componentName = APP_NAME_PREFIX + appName + ".Siddhi." + componentType + "." +
                               componentId + "" + "." + metricsType;
        String resolvedSelectWorkerMetricsHistoryQuery = resolveTableName(selectAppComponentHistory, tableName);
        String resolvedQuery = resolvedSelectWorkerMetricsHistoryQuery.replace(SQLConstants.PLACEHOLDER_BEGIN_TIME,
                QUESTION_MARK).replace(PLACEHOLDER_NAME, QUESTION_MARK).replace
                (PLACEHOLDER_WORKER_ID, QUESTION_MARK).replace(SQLConstants.PLACEHOLDER_CURRENT_TIME,
                QUESTION_MARK).replace(PLACEHOLDER_COLUMNS, tableColumn.get(tableName));
        String[] parameters = new String[] {workerId, componentName + PERCENTAGE_MARK,
                Long.toString(currentTimeMilli - timeInterval), Long.toString(currentTimeMilli)};
        return select(resolvedQuery, tableColumn.get(tableName), tableName, parameters);
    }

    /**
     * Select the component History .
     *
     * @param workerId         ID of the worker
     * @param appName          siddhi application name
     * @param timeInterval     time interval that needed to be taken.
     * @param currentTimeMilli current time in milliseconds expression in db type
     * @param metricsType      table name to be fetched
     * @return
     */
    public List<List<Object>> selectAppComponentsAggHistory(String workerId, String appName, long timeInterval, long
            currentTimeMilli, String metricsType, String componentType, String componentId) {
        long aggregationTime = DBTableUtils.getAggregation(timeInterval);
        Map<String, String> typeTableColumn = DBTableUtils.getInstance().loadMetricsTypeSelection();
        String tableName = typeTableColumn.get(metricsType);
        Map<String, String> tableAggColumn = DBTableUtils.getInstance().loadAggMetricsAllValueSelection();
        Map<String, String> tableColumn = DBTableUtils.getInstance().loadAggRowMetricsAllValueSelection();
        String componentName = "org.wso2.siddhi.SiddhiApps." + appName + ".Siddhi." + componentType + "." + componentId + "" +
                "." + metricsType;
        String resolvedSelectWorkerMetricsHistoryQuery = resolveTableName(selectAppComponentAggregatedHistory, tableName);
        String resolvedQuery = resolvedSelectWorkerMetricsHistoryQuery.replace(SQLConstants.PLACEHOLDER_BEGIN_TIME,
                QUESTION_MARK).replace(PLACEHOLDER_NAME, QUESTION_MARK).replace
                (PLACEHOLDER_WORKER_ID, QUESTION_MARK).replace(SQLConstants.PLACEHOLDER_CURRENT_TIME,
                QUESTION_MARK)
                .replace(PLACEHOLDER_AGGREGATION_TIME, Long.toString(aggregationTime))
                .replace(PLACEHOLDER_AGGREGATION_COMPONENT_COLOUM, tableAggColumn.get(tableName));
        String[] parameters = new String[] {workerId, componentName + PERCENTAGE_MARK,
                Long.toString(currentTimeMilli - timeInterval), Long.toString(currentTimeMilli)};
        return select(resolvedQuery, tableColumn.get(tableName), tableName,parameters);
    }
    /**
     * Select the component list of the siddhi app.
     *
     * @param workerId         ID of the worker
     * @param appName          siddhi application name
     * @param timeInterval     time interval that needed to be taken.
     * @param currentTimeMilli current time in milliseconds expression in db type
     * @return component List
     */
    public Map<String, List<String>> selectAppComponentsList(String workerId, String appName, int timeInterval, long
            currentTimeMilli) {
        Map<String, List<String>> allComponentsNames = new LinkedHashMap<>();
        for (String tableName : METRICS_TABLE_NAMES) {
            List<String> subComponentsList = new ArrayList<>();
            String resolvedSelectWorkerMetricsQuery = resolveTableName(selectAppComponentList, tableName);
            String resolvedQuery = resolvedSelectWorkerMetricsQuery.replace(SQLConstants.PLACEHOLDER_BEGIN_TIME,
                    QUESTION_MARK).replace(PLACEHOLDER_NAME, QUESTION_MARK).replace
                    (PLACEHOLDER_WORKER_ID, QUESTION_MARK).replace(SQLConstants.PLACEHOLDER_CURRENT_TIME,
                    QUESTION_MARK);
            String[] parameters = new String[] {workerId, APP_NAME_PREFIX + appName +"." +PERCENTAGE_MARK,
                    Long.toString(currentTimeMilli - timeInterval), Long.toString(currentTimeMilli)};
            List<List<Object>> list = select(resolvedQuery, "NAME", tableName, parameters);
            if (!list.isEmpty()) {
                for (Object app : list) {
                    subComponentsList.add((String) ((ArrayList) app).get(0));
                }
            }
            for (String componentMetricsName : subComponentsList) {
                String componentName = componentMetricsName.substring(0, componentMetricsName.lastIndexOf("."));
                List<String> existingList = allComponentsNames.get(componentName);
                if (existingList == null) {
                    List<String> newList = new ArrayList<>();
                    newList.add(tableName);
                    allComponentsNames.put(componentName, newList);
                } else {
                    existingList.add(tableName);
                    allComponentsNames.put(componentName, existingList);
                }
            }
        }
        return allComponentsNames;
    }

    /**
     * Select the last Metrics value of components in a particular table.
     *
     * @param carbonId   carbon id of a worker.
     * @param components component list of the worker.
     * @return
     */
    public List selectComponentsLastMetric(String carbonId, String appName, Map<String, List<String>> components,
            long timeInterval, long currentTimeMilli) {
        Map<String, String> tableMetricsMap = new HashMap<>();
        for (Map.Entry<String, String> entry : DBTableUtils.getInstance().loadMetricsTypeSelection().entrySet()) {
            tableMetricsMap.put(entry.getValue(), entry.getKey());
        }
        Map<String, String> tableColumn = DBTableUtils.getInstance().loadMetricsValueSelection();
        List<TypeMetrics> componentsRecentMetrics = new ArrayList<>();
        MetricElement metricElement = new MetricElement();
        ComponentMetrics componentMetrics = new ComponentMetrics();
        TypeMetrics typeMetrics = new TypeMetrics();
        for (Map.Entry componentEntry : components.entrySet()) {
            List<String> componentsTableList = (List<String>) componentEntry.getValue();
            for (String tableEntry : componentsTableList) {
                String columnListString = tableColumn.get(tableEntry);
                String resolvedSelectWorkerMetricsQuery = resolveTableName(selectAppComponentMetrics, tableEntry);
                String resolvedSelectWorkerRecentMetricsQuery = resolveTableName(selectAppComponentHistory,
                        tableEntry);
                String resolvedQuery = resolvedSelectWorkerMetricsQuery.replace(PLACEHOLDER_NAME,
                        QUESTION_MARK).replace(PLACEHOLDER_WORKER_ID, QUESTION_MARK).replace
                        (PLACEHOLDER_COLUMNS, columnListString);
                String resolvedRecentQuery = resolvedSelectWorkerRecentMetricsQuery
                        .replace(PLACEHOLDER_NAME, QUESTION_MARK).replace(PLACEHOLDER_WORKER_ID, QUESTION_MARK)
                        .replace(PLACEHOLDER_COLUMNS, columnListString)
                        .replace(SQLConstants.PLACEHOLDER_BEGIN_TIME, QUESTION_MARK)
                        .replace(SQLConstants.PLACEHOLDER_CURRENT_TIME, QUESTION_MARK);
                String[] recentQueryParameters = new String[] {carbonId,
                        String.format ("%s%s", (String) componentEntry.getKey(), PERCENTAGE_MARK),
                        Long.toString(currentTimeMilli - timeInterval), Long.toString(currentTimeMilli)};
                //String[] resolvedQueryParameters = new String[] {carbonId, (String) componentEntry.getKey()};
                String[] columnList = columnListString.split(",");
                List<List<Object>> selectionRecent =
                        select(resolvedRecentQuery, columnListString, tableEntry, recentQueryParameters);
                List<Object> selection = new ArrayList<>();
                if(selectionRecent.size()>0) {
                    selection = selectionRecent.get(selectionRecent.size() - 1);
                }
                String[] componentElements = ((String) componentEntry.getKey()).replace("org.wso2.siddhi" +
                        ".SiddhiApps" +
                        "." + appName + ".Siddhi.", "").split("\\.", 2);
                String metricType = tableMetricsMap.get(tableEntry).toLowerCase();
                if ((selection != null) && (!selection.isEmpty())) {
                    Attribute attribute = new Attribute(columnList[1], selection.get(1));
                    attribute.setRecentValues(selectionRecent);
                    metricElement.addAttributes(attribute);
                    metricElement.setType(metricType);
                    componentMetrics.addMetrics(metricElement);
                    metricElement = new MetricElement();
                    componentMetrics.setName(componentElements[1]);
                    typeMetrics.setType(componentElements[0]);
                    typeMetrics.setData(componentMetrics);
                }
            }
            boolean isNew = true;
            for (TypeMetrics typeMetric : componentsRecentMetrics) {
                if (typeMetrics.getType().equalsIgnoreCase(typeMetric.getType())) {
                    isNew = false;
                    typeMetric.getData().add(typeMetrics.getData().get(0));
                }
            }
            if (isNew) {
                componentsRecentMetrics.add(typeMetrics);
            }
            componentMetrics = new ComponentMetrics();
            typeMetrics = new TypeMetrics();
        }
        return componentsRecentMetrics;
    }


    /**
     * This method resold the MetricElement query by replacing the values.
     *
     * @param workerId     workerID
     * @param timeInterval timeInterval
     * @param appName      siddhi app name
     * @param currentTime  current time expression.
     * @return selected list of metrics.
     */
    public List selectAppOverallMetrics(String metricsType, String workerId, long
            timeInterval, String appName, long currentTime) {
        switch (metricsType) {
            case "memory": {
                String tableName = "METRIC_GAUGE";
                String columnsOrSelectExpressions = "AVG(CAST(result.VALUE as DECIMAL(22,2)))";
                String resultLabel = "VALUE";
                String resolvedQueryTable = selectAppMetricsQuery.replace(SQLConstants.PLACEHOLDER_COLUMNS,
                        columnsOrSelectExpressions).replace(SQLConstants.PLACEHOLDER_BEGIN_TIME, QUESTION_MARK)
                        .replace(PLACEHOLDER_NAME, QUESTION_MARK)
                        .replace(PLACEHOLDER_WORKER_ID, QUESTION_MARK)
                        .replace(SQLConstants.PLACEHOLDER_CURRENT_TIME, QUESTION_MARK)
                        .replace(PLACEHOLDER_RESULT, resultLabel)
                        .replace(PLACEHOLDER_TABLE_NAME, tableName);
                String[] parameters = new String[] {workerId, APP_NAME_PREFIX + appName + PERCENTAGE_MARK, Long
                        .toString(currentTime - timeInterval), Long.toString(currentTime)};
                return selectAppMemory(resolvedQueryTable, tableName, parameters,"TIMESTAMP");
            }
            case "throughput": {
                String tableName = "METRIC_METER";
                String columnsLabels = "TIMESTAMP,COUNT";
                String columnsOrSelectExpressions = "AVG(result.COUNT)";
                String resultLabel = "COUNT";
                String resolvedQueryTable = selectAppMetricsQuery.replace(SQLConstants.PLACEHOLDER_COLUMNS,
                        columnsOrSelectExpressions).replace(SQLConstants.PLACEHOLDER_BEGIN_TIME, QUESTION_MARK)
                        .replace(PLACEHOLDER_NAME, QUESTION_MARK).replace(PLACEHOLDER_WORKER_ID, QUESTION_MARK)
                        .replace(SQLConstants.PLACEHOLDER_CURRENT_TIME, QUESTION_MARK)
                        .replace(PLACEHOLDER_RESULT, resultLabel)
                        .replace(PLACEHOLDER_TABLE_NAME, tableName);
                String[] parameters = new String[] {workerId, APP_NAME_PREFIX + appName + PERCENTAGE_MARK, Long
                        .toString(currentTime - timeInterval), Long.toString(currentTime)};
                return select(resolvedQueryTable, columnsLabels, tableName, parameters);
            }
            case "latency": {
                String tableName = "METRIC_TIMER";
                String columnsLabels = "TIMESTAMP,COUNT";
                String columnsOrSelectExpressions = "AVG(result.COUNT)";
                String resultLabel = "COUNT";
                String resolvedQueryTable = selectAppMetricsQuery.replace(SQLConstants.PLACEHOLDER_COLUMNS,
                        columnsOrSelectExpressions).replace(SQLConstants.PLACEHOLDER_BEGIN_TIME, QUESTION_MARK)
                        .replace(PLACEHOLDER_NAME, QUESTION_MARK).replace(PLACEHOLDER_WORKER_ID, QUESTION_MARK)
                        .replace(SQLConstants.PLACEHOLDER_CURRENT_TIME, QUESTION_MARK)
                        .replace(PLACEHOLDER_RESULT, resultLabel)
                        .replace(PLACEHOLDER_TABLE_NAME, tableName);
                String[] parameters = new String[] {workerId, APP_NAME_PREFIX + appName + PERCENTAGE_MARK,
                        Long.toString(currentTime - timeInterval), Long.toString(currentTime)};
                return select(resolvedQueryTable, columnsLabels, tableName, parameters);
            }
            default: {
                logger.error("Invalid parameters type: " + removeCRLFCharacters(workerId) + ":"
                        + removeCRLFCharacters(appName));
                return null;
            }
        }
    }
    /**
     * This method resold the MetricElement query by replacing the values.
     *
     * @param workerId     workerID
     * @param timeInterval timeInterval
     * @param appName      siddhi app name
     * @param currentTime  current time expression.
     * @return selected list of metrics.
     */
    public List selectAppAggOverallMetrics(String metricsType, String workerId, long
            timeInterval, String appName, long currentTime) {
        long aggregationTime = DBTableUtils.getAggregation(timeInterval);
        switch (metricsType) {
            case "memory": {
                String tableName = "METRIC_GAUGE";
                String columnsOrSelectExpressions = "AVG(CAST(result.VALUE as DECIMAL(22,2)))";
                String resultLabel = "VALUE";
                String resolvedQueryTable = recordSelectAgregatedAppMetricsQuery.replace(SQLConstants.PLACEHOLDER_COLUMNS,
                        columnsOrSelectExpressions).replace(SQLConstants.PLACEHOLDER_BEGIN_TIME, QUESTION_MARK)
                        .replace(PLACEHOLDER_NAME, QUESTION_MARK)
                        .replace(PLACEHOLDER_WORKER_ID, QUESTION_MARK)
                        .replace(SQLConstants.PLACEHOLDER_CURRENT_TIME, QUESTION_MARK)
                        .replace(PLACEHOLDER_RESULT, resultLabel)
                        .replace(PLACEHOLDER_TABLE_NAME, tableName)
                        .replace(PLACEHOLDER_AGGREGATION_TIME, Long.toString(aggregationTime));
                String[] parameters = new String[] {workerId, APP_NAME_PREFIX + appName + PERCENTAGE_MARK, Long
                        .toString(currentTime - timeInterval), Long.toString(currentTime)};
                return selectAppMemory(resolvedQueryTable, tableName,parameters,"AGG_TIMESTAMP");
            }
            case "throughput": {
                String tableName = "METRIC_METER";
                String columnsLabels = "AGG_TIMESTAMP,COUNT";
                String columnsOrSelectExpressions = "AVG(result.COUNT)";
                String resultLabel = "COUNT";
                String resolvedQueryTable = recordSelectAgregatedAppMetricsQuery.replace(SQLConstants.PLACEHOLDER_COLUMNS,
                        columnsOrSelectExpressions).replace(SQLConstants.PLACEHOLDER_BEGIN_TIME, QUESTION_MARK)
                        .replace(PLACEHOLDER_NAME, QUESTION_MARK).replace
                        (PLACEHOLDER_WORKER_ID, QUESTION_MARK).replace(SQLConstants.PLACEHOLDER_CURRENT_TIME,
                                QUESTION_MARK).replace(PLACEHOLDER_RESULT, resultLabel)
                        .replace(PLACEHOLDER_TABLE_NAME, tableName)
                        .replace(PLACEHOLDER_AGGREGATION_TIME, Long.toString(aggregationTime));
                String[] parameters = new String[] {workerId, APP_NAME_PREFIX + appName + PERCENTAGE_MARK, Long
                        .toString(currentTime - timeInterval), Long.toString(currentTime)};
                return select(resolvedQueryTable, columnsLabels, tableName,parameters);
            }
            case "latency": {
                String tableName = "METRIC_TIMER";
                String columnsLabels = "AGG_TIMESTAMP,COUNT";
                String columnsOrSelectExpressions = "AVG(result.COUNT)";
                String resultLabel = "COUNT";
                String resolvedQueryTable = recordSelectAgregatedAppMetricsQuery.replace(SQLConstants.PLACEHOLDER_COLUMNS,
                        columnsOrSelectExpressions).replace(SQLConstants.PLACEHOLDER_BEGIN_TIME, QUESTION_MARK)
                        .replace(PLACEHOLDER_NAME, QUESTION_MARK).replace
                        (PLACEHOLDER_WORKER_ID, QUESTION_MARK).replace(SQLConstants.PLACEHOLDER_CURRENT_TIME,
                                QUESTION_MARK).replace(PLACEHOLDER_RESULT, resultLabel)
                        .replace(PLACEHOLDER_TABLE_NAME, tableName)
                        .replace(PLACEHOLDER_AGGREGATION_TIME, Long.toString(aggregationTime));
                String[] parameters = new String[] {workerId, APP_NAME_PREFIX + appName + PERCENTAGE_MARK,
                        Long.toString(currentTime - timeInterval), Long.toString(currentTime)};
                return select(resolvedQueryTable, columnsLabels, tableName,parameters);
            }
            default: {
                logger.error("Invalid parameters type: " + removeCRLFCharacters(workerId) + ":"
                        + removeCRLFCharacters(appName));
                return null;
            }
        }
    }

    /**
     * Used to get the metrics gauges of jvm metrics.
     *
     * @param workerId       source ID of the metrics.
     * @param timeInterval   time interval that needed to be taken.
     * @param metricTypeName metrics type name ex: memory,cpu
     * @param currentTime    current time in milliseconds.
     * @return List<List<Object>> of metrics data because charts needed in that format
     */
    public List selectWorkerMetrics(String workerId, long timeInterval, String metricTypeName, long
            currentTime) {
        String resolvedSelectWorkerMetricsQuery = resolveTableName(selectWorkerMetricsQuery, "METRIC_GAUGE");
        String resolvedQuery = resolvedSelectWorkerMetricsQuery.replace(SQLConstants.PLACEHOLDER_BEGIN_TIME,
                QUESTION_MARK).replace(PLACEHOLDER_NAME, QUESTION_MARK)
                .replace(PLACEHOLDER_WORKER_ID, QUESTION_MARK)
                .replace(SQLConstants.PLACEHOLDER_CURRENT_TIME, QUESTION_MARK);
        String[] parameters = new String[] {workerId, Long.toString(currentTime - timeInterval),
                Long.toString(currentTime), metricTypeName};
        return selectGauge(resolvedQuery, false, parameters);
    }

    /**
     * Used to get the metrics gauges of jvm metrics.
     *
     * @param workerId       source ID of the metrics.
     * @param timeInterval   time interval that needed to be taken.
     * @param metricTypeName metrics type name ex: memory,cpu
     * @param currentTime    current time in milliseconds.
     * @return List<List<Object>> of metrics data because charts needed in that format
     */
    public List selectWorkerAggregatedMetrics(String workerId, long timeInterval, String metricTypeName, long
            currentTime) {
        long aggregationTime = DBTableUtils.getAggregation(timeInterval);
        String resolvedSelectWorkerMetricsQuery = resolveTableName(selectWorkerAggregatedMetricsQuery, "METRIC_GAUGE");
        String resolvedQuery = resolvedSelectWorkerMetricsQuery.replace(SQLConstants.PLACEHOLDER_BEGIN_TIME,
                QUESTION_MARK).replace(PLACEHOLDER_NAME, QUESTION_MARK)
                .replace(PLACEHOLDER_WORKER_ID, QUESTION_MARK)
                .replace(SQLConstants.PLACEHOLDER_CURRENT_TIME, QUESTION_MARK)
                .replace(PLACEHOLDER_AGGREGATION_TIME, Long.toString(aggregationTime));
        String[] parameters = new String[] {workerId, Long.toString(currentTime - timeInterval),
                Long.toString(currentTime), metricTypeName};
        return selectGauge(resolvedQuery, true, parameters);
    }

    /**
     * Used to ge the overall throughput of the worker.
     *
     * @param workerId     source id of the metrics
     * @param timeInterval time interval that metrics needed to be taken.
     * @param currentTime  current time
     * @return List<List<Object>> of metrics data because charts needed in that format
     */
    public List selectWorkerThroughput(String workerId, long timeInterval, long currentTime) {
        String resolvedSelectWorkerThroughputQuery = resolveTableName(selectWorkerThroughputQuery,
                "METRIC_METER");
        String resolvedQuery = resolvedSelectWorkerThroughputQuery.replace(SQLConstants.PLACEHOLDER_COLUMNS,
                "AVG(result.COUNT)").replace
                (SQLConstants.PLACEHOLDER_BEGIN_TIME, QUESTION_MARK).replace
                (PLACEHOLDER_WORKER_ID, QUESTION_MARK).replace(SQLConstants.PLACEHOLDER_CURRENT_TIME,
                QUESTION_MARK).replace(PLACEHOLDER_RESULT, "COUNT");
        String[] parameters = new String[] {workerId, Long.toString(currentTime - timeInterval),
                Long.toString(currentTime)};
        return select(resolvedQuery, "TIMESTAMP,COUNT", "METRIC_METER", parameters);
    }

    /**
     * Used to ge the overall throughput of the worker.
     *
     * @param workerId     source id of the metrics
     * @param timeInterval time interval that metrics needed to be taken.
     * @param currentTime  current time
     * @return List<List<Object>> of metrics data because charts needed in that format
     */
    public List selectWorkerAggregatedThroughput(String workerId, long timeInterval, long currentTime) {
        long aggregationTime = DBTableUtils.getAggregation(timeInterval);
        String resolvedSelectWorkerThroughputQuery = resolveTableName(selectWorkerAggregatedThroughputQuery,
                "METRIC_METER");
        String resolvedQuery = resolvedSelectWorkerThroughputQuery.replace(SQLConstants.PLACEHOLDER_COLUMNS,
                "AVG(result.COUNT)").replace
                (SQLConstants.PLACEHOLDER_BEGIN_TIME, QUESTION_MARK).replace
                (PLACEHOLDER_WORKER_ID, QUESTION_MARK).replace(SQLConstants.PLACEHOLDER_CURRENT_TIME,
                QUESTION_MARK).replace(PLACEHOLDER_RESULT, "COUNT").
                replace(PLACEHOLDER_AGGREGATION_TIME, Long.toString(aggregationTime));
        String[] parameters = new String[] {workerId, Long.toString(currentTime - timeInterval),
                Long.toString(currentTime)};
        return select(resolvedQuery, "AGG_TIMESTAMP,COUNT", "METRIC_METER", parameters);
    }

    /**
     * Select the metrics of the siddhi app.
     *
     * @param query selection query.
     * @return the selected object.
     */
    private List<List<Object>> selectAppMemory(String query, String tableName, String[] parameters,String timesStampLable) {
        Map<String, String> attributesTypeMap = workerAttributeTypeMap.get(tableName);
        Connection conn = this.getConnection();
        ResultSet rs = null;
        List<List<Object>> tuple = new ArrayList<>();
        List<Object> row;
        PreparedStatement stmt = null;
        try {
            stmt = conn.prepareStatement(query);
            for (int i = 1; i <= parameters.length; i++) {
                stmt.setString(i, parameters[i-1]);
            }
            rs = DBHandler.getInstance().select(stmt);
            while (rs.next()) {
                row = new ArrayList<>();
                row.add(DBTableUtils.getInstance().fetchData(rs, timesStampLable, attributesTypeMap.get
                        (timesStampLable),metricsQueryManager));
                row.add(Double.valueOf((String) DBTableUtils.getInstance().fetchData(rs, "VALUE",
                        attributesTypeMap.get("VALUE"),metricsQueryManager)));
                tuple.add(row);
            }
        } catch (SQLException e) {
            throw new RDBMSTableException("Error retrieving records from table '" + tableName + "': "
                    + e.getMessage() + " in " + DATASOURCE_ID, e);
        } finally {
            try {
                if (rs != null) {
                    rs.close();
                }
            } catch (SQLException e) {
                //ignore
            }
            try {
                if (stmt != null) {
                    stmt.close();
                }
            } catch (SQLException e) {
                //ignore
            }
        }
        return tuple;
    }

    /**
     * Select the metrics.
     *
     * @param query   selection query.
     * @param columns column labeles that needed to select.
     * @return the selected object.
     */
    private List<List<Object>> select(String query, String columns, String tableName, String[] parameters) {
        Map<String, String> attributesTypeMap = workerAttributeTypeMap.get(tableName);
        Connection conn = this.getConnection();
        ResultSet rs = null;
        List<List<Object>> tuple = new ArrayList<>();
        List<Object> row;
        PreparedStatement stmt = null;
        String[] columnLabels = columns.split(",");
        try {
            stmt = conn.prepareStatement(query);
            for (int i = 1; i <= parameters.length; i++) {
                stmt.setString(i, parameters[i-1]);
            }
            rs = DBHandler.getInstance().select(stmt);
            while (rs.next()) {
                row = new ArrayList<>();
                for (String columnLabel : columnLabels) {
                    if (columnLabel.equalsIgnoreCase("VALUE")) {
                        row.add(Double.valueOf((String) DBTableUtils.getInstance().fetchData(rs, columnLabel, attributesTypeMap.get
                                (columnLabel),metricsQueryManager)));
                    } else {
                        row.add(DBTableUtils.getInstance().fetchData(rs, columnLabel, attributesTypeMap.get
                                (columnLabel),metricsQueryManager));
                    }
                }
                tuple.add(row);
            }
        } catch (SQLException e) {
            throw new RDBMSTableException("Error retrieving records from table '" + tableName + "': "
                    + e.getMessage() + " in " + DATASOURCE_ID, e);
        } finally {
            try {
                if (rs != null) {
                    rs.close();
                }
            } catch (SQLException e) {
                //ignore
            }
            try {
                if (stmt != null) {
                    stmt.close();
                }
            } catch (SQLException e) {
                logger.error(e.getMessage(), e);
                //ignore
            }
        }
        return tuple;
    }

    /**
     * Select the metrics.
     *
     * @param query selection query.
     * @return the selected object.
     */
    private List<List<Object>> selectGauge(String query, boolean isAggregated, String[] parameters) {
        Map<String, String> attributesTypeMap = workerAttributeTypeMap.get("METRIC_GAUGE");
        Connection conn = this.getConnection();
        ResultSet rs = null;
        List<List<Object>> tuple = new ArrayList<>();
        PreparedStatement stmt = null;
        List<Object> row;
        try {
            stmt = conn.prepareStatement(query);
            for (int i = 1; i <= parameters.length; i++) {
                stmt.setString(i, parameters[i-1]);
            }
            rs = DBHandler.getInstance().select(stmt);
            String timestampCol = "TIMESTAMP";
            if (isAggregated) {
                timestampCol = "AGG_TIMESTAMP";
            }
            while (rs.next()) {
                row = new ArrayList<>();
                row.add(DBTableUtils.getInstance().fetchData(rs, timestampCol, attributesTypeMap.get
                        ("TIMESTAMP"),metricsQueryManager));
                row.add(Double.valueOf((String) DBTableUtils.getInstance().fetchData(rs, "VALUE",
                        attributesTypeMap.get
                                ("VALUE"),metricsQueryManager)));
                tuple.add(row);
            }
        } catch (SQLException e) {
            throw new RDBMSTableException("Error retrieving records from table '" + "METRIC_GAUGE" + "': "
                    + e.getMessage() + " in " + DATASOURCE_ID, e);
        } finally {

            if (rs != null) {
                try {
                    rs.close();
                } catch (SQLException e) {
                    //ignore
                }
            }

            if (stmt != null) {
                try {
                    stmt.close();
                } catch (SQLException e) {
                    //ignore
                }
            }

        }
        return tuple;
    }

    /**
     * clean up the database connection.
     */
    public void cleanupConnection() {
        Connection conn = getConnection();
        if (conn != null) {
            try {
                conn.close();
            } catch (SQLException ignore) { /* ignore */ }
        }
    }

    private static String removeCRLFCharacters(String str) {
        if (str != null) {
            str = str.replace('\n', '_').replace('\r', '_');
        }
        return str;
    }
}
