/*
 * 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.hadoop.chukwa.database;


import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.chukwa.inputtools.mdl.DataConfig;
import org.apache.hadoop.chukwa.util.DatabaseWriter;
import org.apache.hadoop.chukwa.util.ExceptionUtil;
import org.apache.hadoop.chukwa.util.PidFile;
import java.sql.SQLException;
import java.sql.ResultSet;
import java.util.Calendar;
import java.util.HashMap;
import java.util.Iterator;
import java.sql.ResultSetMetaData;
import java.text.SimpleDateFormat;

public class Consolidator extends Thread {
  private DatabaseConfig dbc = new DatabaseConfig();

  private static Log log = LogFactory.getLog(Consolidator.class);
  private String table = null;
  private int[] intervals;
  private static PidFile loader = null;

  public Consolidator(String table, String intervalString) {
    super(table);
    try {
      int i = 0;
      String[] temp = intervalString.split("\\,");
      intervals = new int[temp.length];
      for (String s : temp) {
        intervals[i] = Integer.parseInt(s);
        i++;
      }
      this.table = table;
    } catch (NumberFormatException ex) {
      log.error("Unable to parse summary interval");
    }
  }

  public void run() {
    ResultSet rs = null;
    String[] columns;
    int[] columnsType;
    String groupBy = "";

    for (int interval : intervals) {
      // Start reducing from beginning of time;
      Calendar aYearAgo = Calendar.getInstance();
      aYearAgo.set(2008, 1, 1, 0, 0, 0);

      long start = aYearAgo.getTimeInMillis(); // starting from 2008/01/01
      long end = start + (interval * 60000);
      log.debug("start time: " + start);
      log.debug("end time: " + end);
      Calendar now = Calendar.getInstance();
      String cluster = System.getProperty("CLUSTER");
      if (cluster == null) {
        cluster = "unknown";
      }
      DatabaseWriter db = new DatabaseWriter(cluster);
      String fields = null;
      String dateclause = null;
      boolean emptyPrimeKey = false;
      log.info("Consolidate for " + interval + " minutes interval.");

      String[] tmpTable = dbc.findTableName(this.table, start, end);
      String table = tmpTable[0];
      String sumTable = "";
      if (interval == 5) {
        long partition = now.getTime().getTime() / DatabaseConfig.WEEK;
        StringBuilder stringBuilder = new StringBuilder();
        stringBuilder.append(this.table);
        stringBuilder.append("_");
        stringBuilder.append(partition);
        stringBuilder.append("_week");
        table = stringBuilder.toString();
        long partition2 = now.getTime().getTime() / DatabaseConfig.MONTH;
        sumTable = this.table + "_" + partition2 + "_month";
      } else if (interval == 30) {
        long partition = now.getTime().getTime() / DatabaseConfig.MONTH;
        table = this.table + "_" + partition + "_month";
        long partition2 = now.getTime().getTime() / DatabaseConfig.QUARTER;
        sumTable = this.table + "_" + partition2 + "_month";
      } else if (interval == 180) {
        long partition = now.getTime().getTime() / DatabaseConfig.QUARTER;
        table = this.table + "_" + partition + "_quarter";
        long partition2 = now.getTime().getTime() / DatabaseConfig.YEAR;
        sumTable = this.table + "_" + partition2 + "_month";
      } else if (interval == 720) {
        long partition = now.getTime().getTime() / DatabaseConfig.YEAR;
        table = this.table + "_" + partition + "_year";
        long partition2 = now.getTime().getTime() / DatabaseConfig.DECADE;
        sumTable = this.table + "_" + partition2 + "_month";
      }
      // Find the most recent entry
      try {
        String query = "select * from " + sumTable
            + " order by timestamp desc limit 1";
        log.debug("Query: " + query);
        rs = db.query(query);
        if (rs == null) {
          throw new SQLException("Table "+ sumTable + " is undefined.");
        }
        ResultSetMetaData rmeta = rs.getMetaData();
        boolean empty = true;
        if (rs.next()) {
          for (int i = 1; i <= rmeta.getColumnCount(); i++) {
            if (rmeta.getColumnName(i).toLowerCase().equals("timestamp")) {
              start = rs.getTimestamp(i).getTime();
            }
          }
          empty = false;
        }
        if (empty) {
          throw new SQLException("Table is empty.");
        }
        end = start + (interval * 60000);
      } catch (SQLException ex) {
        try {
          String query = "select * from " + table
              + " order by timestamp limit 1";
          log.debug("Query: " + query);
          rs = db.query(query);
          if (rs.next()) {
            ResultSetMetaData rmeta = rs.getMetaData();
            for (int i = 1; i <= rmeta.getColumnCount(); i++) {
              if (rmeta.getColumnName(i).toLowerCase().equals("timestamp")) {
                start = rs.getTimestamp(i).getTime();
              }
            }
          }
          end = start + (interval * 60000);
        } catch (SQLException ex2) {
          log.error("Unable to determine starting point in table: "
              + this.table);
          log.error("SQL Error:" + ExceptionUtil.getStackTrace(ex2));
          return;
        }
      }
      try {
        ResultSetMetaData rmeta = rs.getMetaData();
        int col = rmeta.getColumnCount();
        columns = new String[col];
        columnsType = new int[col];
        for (int i = 1; i <= col; i++) {
          columns[i - 1] = rmeta.getColumnName(i);
          columnsType[i - 1] = rmeta.getColumnType(i);
        }

        for (int i = 0; i < columns.length; i++) {
          if (i == 0) {
            fields = columns[i];
            if (columnsType[i] == java.sql.Types.VARCHAR) {
              if (groupBy.equals("")) {
                groupBy = " group by " + columns[i];
              } else {
                groupBy = groupBy + "," + columns[i];
              }
            }
          } else {
            if (columnsType[i] == java.sql.Types.VARCHAR
                || columnsType[i] == java.sql.Types.TIMESTAMP) {
              fields = fields + "," + columns[i];
              if (columnsType[i] == java.sql.Types.VARCHAR) {
                if (groupBy.equals("")) {
                  groupBy = " group by " + columns[i];
                } else {
                  groupBy = groupBy + "," + columns[i];
                }
              }
            } else {
              fields = fields + ",AVG(" + columns[i] + ") as " + columns[i];
            }
          }
        }
      } catch (SQLException ex) {
        log.error("SQL Error:" + ExceptionUtil.getStackTrace(ex));
        return;
      }
      if (groupBy.equals("")) {
        emptyPrimeKey = true;
      }
      long previousStart = start;
      long partition = 0;
      String timeWindowType = "week";
      while (end < now.getTimeInMillis() - (interval * 2 * 60000)) {
        // Select new data sample for the given intervals
        if (interval == 5) {
          timeWindowType = "month";
          partition = start / DatabaseConfig.MONTH;
        } else if (interval == 30) {
          timeWindowType = "quarter";
          partition = start / DatabaseConfig.QUARTER;
        } else if (interval == 180) {
          timeWindowType = "year";
          partition = start / DatabaseConfig.YEAR;
        } else if (interval == 720) {
          timeWindowType = "decade";
          partition = start / DatabaseConfig.DECADE;
        }
        SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        String startS = formatter.format(start);
        String endS = formatter.format(end);
        dateclause = "Timestamp >= '" + startS + "' and Timestamp <= '" + endS
            + "'";
        if (emptyPrimeKey) {
          groupBy = " group by FLOOR(UNIX_TIMESTAMP(TimeStamp)/" + interval
              * 60 + ")";
        }
        String query = "replace into " + this.table + "_" + partition + "_"
            + timeWindowType + " (select " + fields + " from " + table
            + " where " + dateclause + groupBy + ")";
        log.debug(query);
        db.execute(query);
        if (previousStart == start) {
          start = start + (interval * 60000);
          end = start + (interval * 60000);
          previousStart = start;
        }
      }
      db.close();
    }
  }

  public static void main(String[] args) {
    DataConfig mdl = new DataConfig();
    loader = new PidFile(System.getProperty("CLUSTER") + "Consolidator");
    HashMap<String, String> tableNames = (HashMap<String, String>) mdl
        .startWith("consolidator.table.");
    try {
      Iterator<String> ti = (tableNames.keySet()).iterator();
      while (ti.hasNext()) {
        String table = ti.next();
        String interval = mdl.get(table);
        table = table.substring(19);
        log.info("Summarizing table:" + table);
        Consolidator dbc = new Consolidator(table, interval);
        dbc.run();
      }
    } catch (NullPointerException e) {
      log.error("Unable to summarize database.");
      log.error("Error:" + ExceptionUtil.getStackTrace(e));
    }
    loader.clean();
  }
}