/*
    * 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.hicc;
  
  
  import java.util.TreeMap;
  import java.util.HashMap;
  import java.util.ArrayList;
  import java.util.List;
  import java.sql.*;
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  import org.apache.hadoop.chukwa.util.ExceptionUtil;
  
  public class DatasetMapper {
    private String jdbc;
    private static Log log = LogFactory.getLog(DatasetMapper.class);
    private TreeMap<String, TreeMap<String, Double>> dataset;
    private List<String> labels;
  
    public DatasetMapper(String jdbc) {
      this.jdbc = jdbc;
      this.dataset = new TreeMap<String, TreeMap<String, Double>>();
      this.labels = new ArrayList<String>();
    }
  
    public void execute(String query, boolean groupBySecondColumn,
        boolean calculateSlope, String formatTime, List<Object> parameters) {
      dataset.clear();
      try {
        // The newInstance() call is a work around for some
        // broken Java implementations
        org.apache.hadoop.chukwa.util.DriverManagerUtil.loadDriver().newInstance();
      } catch (Exception ex) {
        log.error("failed to load driver", ex);
        // handle the error
      }
      Connection conn = null;
      PreparedStatement stmt = null;
      ResultSet rs = null;
      labels.clear();
      double max = 0.0;
      long timeWindowSize=0;
      long previousTime=0;
      try {
        conn = org.apache.hadoop.chukwa.util.DriverManagerUtil.getConnection(jdbc);
        stmt = conn.prepareStatement(query);
        if(query.indexOf("?")!=-1) {
          for(int i=0;i<parameters.size();i++) {
            int index = i+1;
            stmt.setObject(index,parameters.get(i));
          }
        }
        // rs = stmt.executeQuery(query);
        if (stmt.execute()) {
          rs = stmt.getResultSet();
          ResultSetMetaData rmeta = rs.getMetaData();
          int col = rmeta.getColumnCount();
          double[] previousArray = new double[col + 1];
          for (int k = 0; k < col; k++) {
            previousArray[k] = 0.0;
          }
          int i = 0;
          java.util.TreeMap<String, Double> data = null;
          HashMap<String, Double> previousHash = new HashMap<String, Double>();
          HashMap<String, Integer> xAxisMap = new HashMap<String, Integer>();
          while (rs.next()) {
            String label = "";
            if (rmeta.getColumnType(1) == java.sql.Types.TIMESTAMP) {
              long time = rs.getTimestamp(1).getTime();
              if(timeWindowSize==0) {
                timeWindowSize=1;
                previousTime=time;
              } else if(time!=previousTime) {
                timeWindowSize=(time-previousTime)/60000;
                previousTime=time;
              }
              label = "" + time;
            } else {
              label = rs.getString(1);
            }
            if (!xAxisMap.containsKey(label)) {
              xAxisMap.put(label, i);
             labels.add(label);
             i++;
           }
           if (groupBySecondColumn) {
             String item = rs.getString(2);
             // Get the data from the row using the series column
             for (int j = 3; j <= col; j++) {
               item = rs.getString(2) + " " + rmeta.getColumnName(j);
               data = dataset.get(item);
               if (data == null) {
                 data = new java.util.TreeMap<String, Double>();
               }
               if (calculateSlope) {
                 double current = rs.getDouble(j);
                 double tmp = 0L;
                 if (data.size() > 1) {
                   tmp = (current - previousHash.get(item).doubleValue())/timeWindowSize;
                   if(timeWindowSize<=0) {
                     tmp = Double.NaN;
                   }
                 } else {
                   tmp = 0;
                 }
                 if (tmp < 0) {
                   tmp = Double.NaN;
                 }
                 previousHash.put(item, current);
                 if (tmp > max) {
                   max = tmp;
                 }
                 data.put(label, tmp);
               } else {
                 double current = rs.getDouble(j);
                 if (current > max) {
                   max = current;
                 }
                 data.put(label, current);
               }
               dataset.put(item, data);
             }
           } else {
             for (int j = 2; j <= col; j++) {
               String item = rmeta.getColumnName(j);
               // Get the data from the row using the column name
               double current = rs.getDouble(j);
               if (current > max) {
                 max = current;
               }
               data = dataset.get(item);
               if (data == null) {
                 data = new java.util.TreeMap<String, Double>();
               }
               if (calculateSlope) {
                 double tmp = current;
                 if (data.size() > 1) {
                   tmp = (tmp - previousArray[j])/timeWindowSize;
                   if(timeWindowSize<=0) {
                     tmp = Double.NaN;
                   }
                 } else {
                   tmp = 0.0;
                 }
                 if (tmp < 0) {
                   tmp = Double.NaN;
                 }
                 previousArray[j] = current;
                 data.put(label, tmp);
               } else {
                 data.put(label, current);
               }
               dataset.put(item, data);
             }
           }
         }
       } else {
         log.error("query is not executed.");
       }
       // Now do something with the ResultSet ....
     } catch (SQLException ex) {
       // handle any errors
       log.error("SQLException: " + ex.getMessage() + " on query: " + query);
       log.error("SQLState: " + ex.getSQLState());
       log.error("VendorError: " + ex.getErrorCode());
     } catch (Exception ex) {
       log.debug(ExceptionUtil.getStackTrace(ex));
     } finally {
       // it is a good idea to release
       // resources in a finally{} block
       // in reverse-order of their creation
       // if they are no-longer needed
       if (rs != null) {
         try {
           rs.close();
         } catch (SQLException sqlEx) {
           log.debug(ExceptionUtil.getStackTrace(sqlEx));
         }
         rs = null;
       }
       if (stmt != null) {
         try {
           stmt.close();
         } catch (SQLException sqlEx) {
           log.debug(ExceptionUtil.getStackTrace(sqlEx));
         }
         stmt = null;
       }
       if (conn != null) {
         try {
           conn.close();
         } catch (SQLException sqlEx) {
           log.debug(ExceptionUtil.getStackTrace(sqlEx));
         }
         conn = null;
       }
     }
   }
 
   public List<String> getXAxisMap() {
     return labels;
   }
 
   public TreeMap<String, TreeMap<String, Double>> getDataset() {
     return dataset;
   }
 }