/*
 * 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.analysis.salsa.visualization;

import prefuse.data.io.sql.*;
import prefuse.data.expression.parser.*;
import prefuse.data.expression.*;
import prefuse.data.column.*;
import prefuse.data.query.*;
import prefuse.data.*;
import prefuse.action.*;
import prefuse.action.layout.*;
import prefuse.action.assignment.*;
import prefuse.visual.expression.*;
import prefuse.visual.*;
import prefuse.render.*;
import prefuse.util.collections.*;
import prefuse.util.*;
import prefuse.*;

import org.apache.hadoop.chukwa.hicc.OfflineTimeHandler;
import org.apache.hadoop.chukwa.hicc.TimeHandler;
import org.apache.hadoop.chukwa.util.DatabaseWriter;
import org.apache.hadoop.chukwa.database.Macro;
import org.apache.hadoop.chukwa.util.XssFilter;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import javax.servlet.http.*;

import java.sql.*;
import java.util.*;

import java.awt.Font;
import java.awt.geom.Rectangle2D;

/**
 * Static image generation for Swimlanes visualization for scalable 
 * rendering on front-end client (web-browser)
 * Handles database data retrieval, transforming data to form for 
 * visualization elements, and initializing and calling visualization
 * elements
 */
public class Swimlanes {

  private static Log log = LogFactory.getLog(Swimlanes.class);

  int SIZE_X=1600, SIZE_Y=1600;
  final int [] BORDER = {50,50,50,50};
  final int LEGEND_X_OFFSET = 50;
  final int LEGEND_Y_OFFSET = 25;
  final int LEGEND_TEXT_OFFSET = 20;
  final int LEGEND_FONT_SIZE = 18;
  final int AXIS_NAME_FONT_SIZE = 24;

  protected boolean offline_use = true;
  protected HttpServletRequest request;
  
  protected String abc;

  /**
   * Modifier for generic Swimlanes plots to plot shuffle, sort, and reducer
   * states of same reduce on same line 
   */
  protected static class MapReduceSwimlanes {
    protected Table plot_tab;
    protected HashMap<String, ArrayList<Tuple> > reducepart_hash;
    protected boolean collate_reduces = false;
    
    public MapReduceSwimlanes() {
      this.plot_tab = new Table();
      this.plot_tab.addColumn("ycoord",float.class);
      this.plot_tab.addColumn("state_name",String.class);
      this.plot_tab.addColumn("hostname",String.class);
      this.plot_tab.addColumn("friendly_id",String.class);
      this.plot_tab.addColumn(START_FIELD_NAME,double.class);
      this.plot_tab.addColumn(END_FIELD_NAME,double.class);
      this.plot_tab.addColumn(PolygonRenderer.POLYGON,float[].class);
      this.reducepart_hash = new HashMap<String, ArrayList<Tuple> >();
    }
    
    public void populateTable_OneLinePerState(Table orig_tab) {
      IntIterator rownumiter;
      int newrownum, origrownum;
      rownumiter = orig_tab.rows(); // iterate over everything
      while (rownumiter.hasNext()) {
        origrownum = ((Integer)rownumiter.next()).intValue();
        newrownum = this.plot_tab.addRow();
        this.plot_tab.set(newrownum, "state_name", orig_tab.getString(origrownum, "state_name"));
        this.plot_tab.set(newrownum, "ycoord", orig_tab.getInt(origrownum, "seqno"));
        this.plot_tab.set(newrownum,"hostname",orig_tab.getString(origrownum,"hostname"));
        this.plot_tab.set(newrownum,"friendly_id",orig_tab.getString(origrownum,"friendly_id"));
        this.plot_tab.set(newrownum,START_FIELD_NAME, orig_tab.getDouble(origrownum,START_FIELD_NAME));
        this.plot_tab.set(newrownum,END_FIELD_NAME, orig_tab.getDouble(origrownum,END_FIELD_NAME));
      }      
    }
    
    public void populateTable_CollateReduces(Table orig_tab) {
      IntIterator rownumiter;
      int newrownum, origrownum;
      
      this.collate_reduces = true;
      
      // add maps normally
      rownumiter = orig_tab.rows(
        (Predicate) ExpressionParser.parse("[state_name] == 'map' " + 
          "OR [state_name] == 'shuffle_local' " + 
          "OR [state_name] == 'shuffle_remote'")
      );
      
      while (rownumiter.hasNext()) {
        origrownum = ((Integer)rownumiter.next()).intValue();
        newrownum = this.plot_tab.addRow();
        this.plot_tab.set(newrownum, "state_name", orig_tab.getString(origrownum, "state_name"));
        this.plot_tab.set(newrownum, "ycoord", orig_tab.getInt(origrownum, "seqno"));
        this.plot_tab.set(newrownum,"hostname",orig_tab.getString(origrownum,"hostname"));
        this.plot_tab.set(newrownum,"friendly_id",orig_tab.getString(origrownum,"friendly_id"));
        this.plot_tab.set(newrownum,START_FIELD_NAME, orig_tab.getDouble(origrownum,START_FIELD_NAME));
        this.plot_tab.set(newrownum,END_FIELD_NAME, orig_tab.getDouble(origrownum,END_FIELD_NAME));
      }

      // special breakdown for reduces
      IntIterator rownumiter3 = orig_tab.rows(
        (Predicate) ExpressionParser.parse("[state_name] == 'reduce_reducer' " +
          "OR [state_name] == 'reduce_shufflewait' " + 
          "OR [state_name] == 'reduce_sort' " + 
          "OR [state_name] == 'reduce'")
      );
      
      ArrayList<Tuple> tuple_array;
      while (rownumiter3.hasNext()) {
        origrownum = ((Integer)rownumiter3.next()).intValue();
        if (orig_tab.getString(origrownum,"state_name").equals("reduce")) {
          continue; // do NOT add reduces
        }
        String curr_reduce = orig_tab.getString(origrownum, "friendly_id");
        newrownum = this.plot_tab.addRow();
        
        this.plot_tab.set(newrownum, "state_name", orig_tab.getString(origrownum, "state_name"));
        this.plot_tab.set(newrownum, "ycoord", orig_tab.getInt(origrownum, "seqno"));
        this.plot_tab.set(newrownum,"hostname",orig_tab.getString(origrownum,"hostname"));
        this.plot_tab.set(newrownum,"friendly_id",orig_tab.getString(origrownum,"friendly_id"));        
        this.plot_tab.set(newrownum,START_FIELD_NAME, orig_tab.getDouble(origrownum,START_FIELD_NAME));
        this.plot_tab.set(newrownum,END_FIELD_NAME, orig_tab.getDouble(origrownum,END_FIELD_NAME));
        
        tuple_array = this.reducepart_hash.get(curr_reduce);
        if (tuple_array == null) {
          tuple_array = new ArrayList<Tuple>();
          tuple_array.add(this.plot_tab.getTuple(newrownum));
          this.reducepart_hash.put(curr_reduce, tuple_array);
        } else {
          tuple_array.add(this.plot_tab.getTuple(newrownum));
        }
      }  
    }
    
    public void populateTable_MapsReducesOnly(Table orig_tab) {
      IntIterator rownumiter;
      int newrownum, origrownum;
      rownumiter = orig_tab.rows(
        (Predicate) ExpressionParser.parse("[state_name] == 'map' OR [state_name] == 'reduce'")
      );
      while (rownumiter.hasNext()) {
        origrownum = ((Integer)rownumiter.next()).intValue();
        newrownum = this.plot_tab.addRow();
        this.plot_tab.set(newrownum, "state_name", orig_tab.getString(origrownum, "state_name"));
        this.plot_tab.set(newrownum, "ycoord", orig_tab.getInt(origrownum, "seqno"));
        this.plot_tab.set(newrownum,"hostname",orig_tab.getString(origrownum,"hostname"));
        this.plot_tab.set(newrownum,"friendly_id",orig_tab.getString(origrownum,"friendly_id"));        
        this.plot_tab.set(newrownum,START_FIELD_NAME, orig_tab.getDouble(origrownum,START_FIELD_NAME));
        this.plot_tab.set(newrownum,START_FIELD_NAME, orig_tab.getDouble(origrownum,END_FIELD_NAME));
      }
    }
    
    /**
     * Reassigns Y coord values to group by state
     */
    public void groupByState() {
      int counter, rownum;
      int rowcount = this.plot_tab.getRowCount();
      HashSet<String> states = new HashSet<String>();
      String curr_state = null;
      Iterator<String> state_iter;
      IntIterator rownumiter;
      
      for (int i = 0; i < rowcount; i++) {
        states.add(this.plot_tab.getString(i,"state_name"));
      }
     
      state_iter = states.iterator();
      counter = 1;
      while (state_iter.hasNext()) {
        curr_state = state_iter.next();
        
        if (this.collate_reduces && ((curr_state.equals("reduce_reducer") || curr_state.equals("reduce_sort")))) {
          continue;
        }
        rownumiter = this.plot_tab.rows(
          (Predicate) ExpressionParser.parse("[state_name] == '"+curr_state+"'")
        );
        if (this.collate_reduces && curr_state.equals("reduce_shufflewait")) {
          while (rownumiter.hasNext()) {
            rownum = ((Integer)rownumiter.next()).intValue();
            this.plot_tab.setFloat(rownum,"ycoord",(float)counter);
            
            ArrayList<Tuple> alt = this.reducepart_hash.get(this.plot_tab.getString(rownum,"friendly_id"));
            Object [] tarr = alt.toArray();
            for (int i = 0; i < tarr.length; i++) ((Tuple)tarr[i]).setFloat("ycoord",(float)counter);
            counter++;            
          }
        } else {
          while (rownumiter.hasNext()) {
            rownum = ((Integer)rownumiter.next()).intValue();
            this.plot_tab.setFloat(rownum,"ycoord",(float)counter);
            counter++;
          }          
        }
      }
    }
    
    public void groupByStartTime() {
      int counter, rownum;
      String curr_state = null;
      IntIterator rownumiter;
     
      rownumiter = this.plot_tab.rowsSortedBy(START_FIELD_NAME, true);
     
      counter = 1;
      while (rownumiter.hasNext()) {
        rownum = ((Integer)rownumiter.next()).intValue();
        curr_state = this.plot_tab.getString(rownum, "state_name");        

        if (this.collate_reduces && curr_state.equals("reduce_shufflewait")) {
          this.plot_tab.setFloat(rownum,"ycoord",(float)counter);
          ArrayList<Tuple> alt = this.reducepart_hash.get(this.plot_tab.getString(rownum,"friendly_id"));
          Object [] tarr = alt.toArray();
          for (int i = 0; i < tarr.length; i++) ((Tuple)tarr[i]).setFloat("ycoord",(float)counter);
          counter++;   
        } else if (!curr_state.equals("reduce_sort") && !curr_state.equals("reduce_reducer")) {
          this.plot_tab.setFloat(rownum,"ycoord",(float)counter);          
          counter++;
        }
      }
    }
    
    public void groupByEndTime() {
      int counter, rownum;
      String curr_state = null;
      IntIterator rownumiter;
     
      rownumiter = this.plot_tab.rowsSortedBy(END_FIELD_NAME, true);
      counter = 1;
      while (rownumiter.hasNext()) {
        rownum = ((Integer)rownumiter.next()).intValue();
        curr_state = this.plot_tab.getString(rownum, "state_name");        

        if (this.collate_reduces && curr_state.equals("reduce_reducer")) {
          this.plot_tab.setFloat(rownum,"ycoord",(float)counter);
          ArrayList<Tuple> alt = this.reducepart_hash.get(this.plot_tab.getString(rownum,"friendly_id"));
          Object [] tarr = alt.toArray();
          for (int i = 0; i < tarr.length; i++) ((Tuple)tarr[i]).setFloat("ycoord",(float)counter);
          counter++;
        } else if (!curr_state.equals("reduce_sort") && !curr_state.equals("reduce_shufflewait")) {
          this.plot_tab.setFloat(rownum,"ycoord",(float)counter);          
          counter++;
        }
      }
    }    
    
    public VisualTable addToVisualization(Visualization viz, String groupname) {
      return viz.addTable(groupname, this.plot_tab);
    }
  }

  /**
   * Provide constant mapping between state names and colours
   * so that even if particular states are missing, the colours are fixed
   * for each state
   */
  public static class SwimlanesStatePalette {
    protected final String [] states = {"map","reduce","reduce_shufflewait","reduce_sort","reduce_reducer","shuffle"};
    HashMap<String,Integer> colourmap; 
    protected int [] palette;
    public SwimlanesStatePalette() {
      palette = ColorLib.getCategoryPalette(states.length);
      colourmap = new HashMap<String,Integer>();
      for (int i = 0; i < states.length; i++) {
        colourmap.put(states[i], Integer.valueOf(palette[i]));
      }
    }
    public int getColour(String state_name) {
      Integer val = colourmap.get(state_name);
      if (val == null) {
        return ColorLib.color(java.awt.Color.BLACK);
      } else {
        return val.intValue();
      }
    }
    public int getNumStates() {
      return this.states.length;
    }
    public String [] getStates() {
      return this.states.clone();
    }
  }

  /**
   * Provides convenient rescaling of raw values to be plotted to
   * actual pixels for plotting on image
   */
  public static class CoordScaler {
    double x_pixel_size, y_pixel_size;
    double x_max_value, y_max_value, x_min_value, y_min_value;
    double x_start, y_start;
    
    public CoordScaler() {
      this.x_pixel_size = 0.0;
      this.y_pixel_size = 0.0;
      this.x_max_value = 1.0;
      this.y_max_value = 1.0;
      this.x_min_value = 0.0;
      this.y_min_value = 0.0;
      this.x_start = 0.0;
      this.y_start = 0.0;
    }
    public void set_pixel_start(double x, double y) {
      this.x_start = x;
      this.y_start = y;
    }
    public void set_pixel_size(double x, double y) {
      this.x_pixel_size = x;
      this.y_pixel_size = y;
    }
    public void set_value_ranges(double x_min, double y_min, double x_max, double y_max) {
      this.x_max_value = x_max;
      this.y_max_value = y_max;
      this.x_min_value = x_min;
      this.y_min_value = y_min;
    }
    public double get_x_coord(double x_value) {
      return x_start+(((x_value - x_min_value) / (x_max_value-x_min_value)) * x_pixel_size);
    }
    public double get_y_coord(double y_value) {
      // this does "inverting" to shift the (0,0) point from top-right to bottom-right
      return y_start+(y_pixel_size - ((((y_value - y_min_value) / (y_max_value-y_min_value)) * y_pixel_size)));
    }
  }

  /**
   * Prefuse action for plotting a line for each state
   */
  public static class SwimlanesStateAction extends GroupAction {
    
    protected CoordScaler cs;
    
    public SwimlanesStateAction() {
      super();
    }
    
    public SwimlanesStateAction(String group, CoordScaler cs) {
      super(group);
      this.cs = cs;
    }
    
    public void run (double frac) {
      VisualItem item = null;
      SwimlanesStatePalette pal = new SwimlanesStatePalette();
      
      Iterator<?> curr_group_items = this.m_vis.items(this.m_group);
          
      while (curr_group_items.hasNext()) {
        item = (VisualItem) curr_group_items.next();
        
        double start_time = item.getDouble(START_FIELD_NAME);
        double finish_time = item.getDouble(END_FIELD_NAME);        
        item.setShape(Constants.POLY_TYPE_LINE);
        item.setX(0.0);
        item.setY(0.0);        
        
        float [] coords = new float[4];
        coords[0] = (float) cs.get_x_coord(start_time);
        coords[1] = (float) cs.get_y_coord((double)item.getInt("ycoord"));
        coords[2] = (float) cs.get_x_coord(finish_time);
        coords[3] = (float) cs.get_y_coord((double)item.getInt("ycoord"));

        item.set(VisualItem.POLYGON,coords);
        item.setStrokeColor(pal.getColour(item.getString("state_name")));
      }
    }    
  } // SwimlanesStateAction

  // keys that need to be filled:
  // period (last1/2/3/6/12/24hr,last7d,last30d), time_type (range/last), start, end
  protected HashMap<String, String> param_map;
  
  protected String cluster;
  protected String timezone;
  protected String shuffle_option;
  protected final String table = "mapreduce_fsm";
  protected boolean plot_legend = true;
  protected String jobname = null;
  
  protected Display dis;
  protected Visualization viz;
  
  protected Rectangle2D dataBound = new Rectangle2D.Double();
  protected Rectangle2D xlabBound = new Rectangle2D.Double();
  protected Rectangle2D ylabBound = new Rectangle2D.Double();
  protected Rectangle2D labelBottomBound = new Rectangle2D.Double();
  
  static final String START_FIELD_NAME = "start_time_num";
  static final String END_FIELD_NAME = "finish_time_num";
  
  /* Different group names allow control of what Renderers to use */
  final String maingroup = "Job";
  final String othergroup = "Misc";
  final String labelgroup = "Label";
  final String legendgroup = "Legend";
  final String legendshapegroup = "LegendShape";
  
  public Swimlanes() {
    this.cluster = "";
    this.timezone = "";
    this.shuffle_option = "";
    param_map = new HashMap<String, String>();
  }
  
  /**
   * @brief Constructor for Swimlanes visualization object
   * @param timezone Timezone string from environment
   * @param cluster Cluster name from environment
   * @param event_type Whether to display shuffles or not
   * @param valmap HashMap of key/value pairs simulating parameters from a HttpRequest
   */
  public Swimlanes
    (String timezone, String cluster, String event_type, 
     HashMap<String, String> valmap) 
  {
    this.cluster = cluster;
    if (timezone != null) {
      this.timezone = timezone;
    } else {
      this.timezone = null;
    }
    this.shuffle_option = event_type;
    
    /* This should "simulate" an HttpServletRequest
     * Need to have "start" and "end" in seconds since Epoch
     */
    this.param_map = valmap;
  }
  
  public Swimlanes
    (String timezone, String cluster, String event_type, 
     HashMap<String, String> valmap, int width, int height) 
  {
    this.cluster = cluster;
    if (timezone != null) {
      this.timezone = timezone;
    } else {
      this.timezone = null;
    }
    this.shuffle_option = event_type;
    
    /* This should "simulate" an HttpServletRequest
     * Need to have "start" and "end" in seconds since Epoch
     */
    this.param_map = valmap; 
    
    this.SIZE_X = width;
    this.SIZE_Y = height;
  }
  
  public Swimlanes
    (String timezone, String cluster, String event_type, 
     HashMap<String, String> valmap, int width, int height,
     String legend_opt) 
  {
    this.cluster = cluster;
    if (timezone != null) {
      this.timezone = timezone;
    } else {
      this.timezone = null;
    }
    this.shuffle_option = event_type;
    
    /* This should "simulate" an HttpServletRequest
     * Need to have "start" and "end" in seconds since Epoch
     */
    this.param_map = valmap;
    
    this.SIZE_X = width;
    this.SIZE_Y = height;
    
    if (legend_opt.equals("nolegend")) {
      this.plot_legend = false;
    }
    
  }
  
  public Swimlanes(HttpServletRequest request) {
    XssFilter xf = new XssFilter(request);
    this.offline_use = false;
    this.request = request;
    HttpSession session = request.getSession();
    this.cluster = session.getAttribute("cluster").toString();
      String evt_type = xf.getParameter("event_type");
    if (evt_type != null) {
      this.shuffle_option = evt_type;
    } else {
      this.shuffle_option = "noshuffle";
    }
    this.timezone = session.getAttribute("time_zone").toString();
  }
  
  /**
   * Set job ID to filter results on
   * Call before calling @see #run
   */
  public void setJobName(String s) {
    this.jobname = s;
  }

  /**
   * Set dimensions of image to be generated
   * Call before calling @see #run
   */  
  public void setDimensions(int width, int height) {
    this.SIZE_X=width;
    this.SIZE_Y=height;
  }
  
  /**
   * Specify whether to print legend of states
   * Advisable to not print legend for excessively small images since
   * legend has fixed point size
   * Call before calling @see #run
   */
  public void setLegend(boolean legendopt) {
    if (legendopt) {
      this.plot_legend = true;
    } else {
      this.plot_legend = false;
    }
  }
  
  /**
   * Generates image in specified format, and writes image as binary
   * output to supplied output stream 
   */
  public boolean getImage(java.io.OutputStream output, String img_fmt, double scale) {
    dis = new Display(this.viz);
    dis.setSize(SIZE_X,SIZE_Y);
    dis.setHighQuality(true);
    dis.setFont(new Font(Font.SANS_SERIF,Font.PLAIN,24));
    return dis.saveImage(output, img_fmt, scale);
  } 
  
  /**
   * Adds a column to given table by converting timestamp to long with
   * seconds since epoch, and adding milliseconds from additional column
   * in original table
   *
   * @param origTable Table to add to
   * @param srcFieldName Name of column containing timestamp
   * @param srcMillisecondFieldName Name of column containing millisecond value of time 
   * @param dstFieldName Name of new column to add
   * 
   * @return Modified table with added column
   */
  protected Table addTimeCol 
    (Table origTable, String srcFieldName, 
     String srcMillisecondFieldName, String dstFieldName)
  {
    origTable.addColumn(dstFieldName, long.class);
    
    int total_rows = origTable.getRowCount();
    for (int curr_row_num = 0; curr_row_num < total_rows; curr_row_num++) {
      origTable.setLong(curr_row_num, dstFieldName, 
        ((Timestamp)origTable.get(curr_row_num, srcFieldName)).getTime() + 
        origTable.getLong(curr_row_num, srcMillisecondFieldName)
      );
    }
    
    return origTable;
  }
  
  /**
   * Adds a column with number of seconds of timestamp elapsed since lowest
   * start time; allows times to be plotted as a delta of the start time
   * 
   * @param origTable Table to add column to
   * @param srcFieldName Name of column containing timestamp
   * @param srcMillisecondFieldName Name of column containing millisecond value of time 
   * @param dstFieldName Name of new column to add
   *   
   * @return Modified table with added column
   */
  protected Table addTimeOffsetCol
    (Table origTable, String srcFieldName,
     String srcMillisecondFieldName, String dstFieldName,
     long timeOffset) 
  {
    Table newtable = addTimeCol(origTable, srcFieldName, 
      srcMillisecondFieldName, dstFieldName + "_fulltime");
      
    ColumnMetadata dstcol = newtable.getMetadata(dstFieldName + "_fulltime");
    long mintime = newtable.getLong(dstcol.getMinimumRow(), dstFieldName + "_fulltime");
    
    if (timeOffset == 0) {
      newtable.addColumn(dstFieldName, "ROUND((["+dstFieldName+"_fulltime] - " + mintime +"L) / 1000L)");
    } else {
      newtable.addColumn(dstFieldName, "ROUND((["+dstFieldName+"_fulltime] - " + timeOffset +"L) / 1000L)");      
    }
    
    return newtable;
  }
  
  protected void setupRenderer() {
    this.viz.setRendererFactory(new RendererFactory(){
      AbstractShapeRenderer sr = new ShapeRenderer();
      ShapeRenderer sr_big = new ShapeRenderer(20);
      Renderer arY = new AxisRenderer(Constants.LEFT, Constants.TOP);
      Renderer arX = new AxisRenderer(Constants.CENTER, Constants.BOTTOM);
      PolygonRenderer pr = new PolygonRenderer(Constants.POLY_TYPE_LINE);
      LabelRenderer lr = new LabelRenderer("label");
      LabelRenderer lr_legend = new LabelRenderer("label");
      
      public Renderer getRenderer(VisualItem item) {
        lr.setHorizontalAlignment(Constants.CENTER);
        lr.setVerticalAlignment(Constants.TOP);
        lr_legend.setHorizontalAlignment(Constants.LEFT);
        lr_legend.setVerticalAlignment(Constants.CENTER);
        
        if (item.isInGroup("ylab")) {
          return arY;
        } else if (item.isInGroup("xlab")) {
          return arX;
        } else if (item.isInGroup(maingroup)) {
          return pr;
        } else if (item.isInGroup(labelgroup)) {
          return lr;
        } else if (item.isInGroup(legendgroup)) {
          return lr_legend;
        } else if (item.isInGroup(legendshapegroup)) {
          return sr_big;
        } else {
          return sr;
        }
      }
    });
  }
  
  // setup columns: add additional time fields
  protected Table setupDataTable() {
    Table res_tab = this.getData();    
    if (res_tab == null) {
        return res_tab;
    }
    
    res_tab.addColumn("seqno","ROW()");
    res_tab = addTimeOffsetCol(res_tab, "start_time", "start_time_millis", START_FIELD_NAME, 0);    
    ColumnMetadata dstcol = res_tab.getMetadata(START_FIELD_NAME);
    long mintime = ((Timestamp)res_tab.get(dstcol.getMinimumRow(), "start_time")).getTime();
    res_tab = addTimeOffsetCol(res_tab, "finish_time", "finish_time_millis", END_FIELD_NAME, mintime);    
    res_tab.addColumn(PolygonRenderer.POLYGON,float[].class);
    
    log.debug("After adding seqno: #cols: " + res_tab.getColumnCount() + "; #rows: " + res_tab.getRowCount());
    
    return res_tab;
  }
  
  protected void addAxisNames() {
    Table textlabels_table = new Table();
    textlabels_table.addColumn("label",String.class);
    textlabels_table.addColumn("type",String.class);
    textlabels_table.addRow();
    textlabels_table.setString(0,"label","Time/s");
    textlabels_table.setString(0,"type","xaxisname");
    
    VisualTable textlabelsviz = this.viz.addTable(labelgroup, textlabels_table);
    textlabelsviz.setX(0,SIZE_X/2d);
    textlabelsviz.setY(0,SIZE_Y - BORDER[2] + (BORDER[2]*0.1));
    textlabelsviz.setTextColor(0,ColorLib.color(java.awt.Color.GRAY));
    textlabelsviz.setFont(0,new Font(Font.SANS_SERIF,Font.PLAIN,AXIS_NAME_FONT_SIZE));
  }
  
  protected void addLegend() {
    SwimlanesStatePalette ssp = new SwimlanesStatePalette();
    
    Table shapes_table = new Table();
    shapes_table.addColumn(VisualItem.X,float.class);
    shapes_table.addColumn(VisualItem.Y,float.class);
    
    Table legend_labels_table = new Table();
    legend_labels_table.addColumn("label",String.class);
    
    // add labels
    int num_states = ssp.getNumStates();
    String [] state_names = ssp.getStates();
    legend_labels_table.addRows(num_states);
    shapes_table.addRows(num_states);
    for (int i = 0; i < num_states; i++) {
      legend_labels_table.setString(i,"label",state_names[i]);
    }
    
    // add legend shapes, manipulate visualitems to set colours
    VisualTable shapes_table_viz = viz.addTable(legendshapegroup, shapes_table);
    float start_x = BORDER[0] + LEGEND_X_OFFSET;
    float start_y = BORDER[1] + LEGEND_Y_OFFSET;
    float incr = (float) 30.0;
    for (int i = 0; i < num_states; i++) {
      shapes_table_viz.setFillColor(i, ssp.getColour(state_names[i]));
      shapes_table_viz.setFloat(i, VisualItem.X, start_x);
      shapes_table_viz.setFloat(i, VisualItem.Y, start_y + (i * incr));
    }
    
    // add legend labels, manipulate visualitems to set font
    VisualTable legend_labels_table_viz = this.viz.addTable(legendgroup, legend_labels_table);
    for (int i = 0; i < num_states; i++) {
      legend_labels_table_viz.setFloat(i, VisualItem.X, start_x + LEGEND_TEXT_OFFSET);
      legend_labels_table_viz.setFloat(i, VisualItem.Y, start_y + (i * incr));
      legend_labels_table_viz.setTextColor(i,ColorLib.color(java.awt.Color.BLACK));
      legend_labels_table_viz.setFont(i,new Font(Font.SANS_SERIF,Font.PLAIN,LEGEND_FONT_SIZE));
    }
    
  }
  
  public void run() {

    // setup bounds
    this.dataBound.setRect(BORDER[0],BORDER[1],SIZE_X-BORDER[2]-BORDER[0],SIZE_Y-BORDER[3]-BORDER[1]);
    this.xlabBound.setRect(BORDER[0],BORDER[1],SIZE_X-BORDER[2]-BORDER[0],SIZE_Y-BORDER[3]-BORDER[1]);
    this.ylabBound.setRect(BORDER[0],BORDER[1],SIZE_X-BORDER[2]-BORDER[0],SIZE_Y-BORDER[3]-BORDER[1]);
    this.labelBottomBound.setRect(BORDER[0],SIZE_X-BORDER[2],SIZE_Y-BORDER[0]-BORDER[1],BORDER[3]);
    
    // setup visualization
    this.viz = new Visualization();
    this.setupRenderer();
    
    // add table to visualization
    Table raw_data_tab = this.setupDataTable();
    MapReduceSwimlanes mrs = new MapReduceSwimlanes();
    mrs.populateTable_CollateReduces(raw_data_tab);
    mrs.groupByState();
    VisualTable maindatatable = mrs.addToVisualization(this.viz, maingroup);
        
    addAxisNames();
    if (plot_legend) {
      addLegend();
    }

    // plot swimlanes lines: setup axes, call custom action
    ActionList draw = new ActionList();
    {
      // setup axes
      AxisLayout xaxis = new AxisLayout(maingroup, START_FIELD_NAME, Constants.X_AXIS, VisiblePredicate.TRUE);
      AxisLayout yaxis = new AxisLayout(maingroup, "ycoord", Constants.Y_AXIS, VisiblePredicate.FALSE);    
      xaxis.setLayoutBounds(dataBound);
      yaxis.setLayoutBounds(dataBound);
    
      ColumnMetadata starttime_meta = maindatatable.getMetadata(START_FIELD_NAME);    
      ColumnMetadata finishtime_meta = maindatatable.getMetadata(END_FIELD_NAME);    
      ColumnMetadata ycoord_meta = maindatatable.getMetadata("ycoord");
      long x_min = (long) ((Double)maindatatable.get(starttime_meta.getMinimumRow(), START_FIELD_NAME)).doubleValue();
      long x_max = (long) ((Double)maindatatable.get(finishtime_meta.getMaximumRow(), END_FIELD_NAME)).doubleValue();
      xaxis.setRangeModel(new NumberRangeModel(x_min,x_max,x_min,x_max));
      float y_max = maindatatable.getFloat(ycoord_meta.getMaximumRow(),"ycoord");
      yaxis.setRangeModel(new NumberRangeModel(0,y_max,0,y_max));
      
      // call custom action to plot actual swimlanes lines
      CoordScaler cs = new CoordScaler();
      cs.set_pixel_size(SIZE_X-BORDER[0]-BORDER[2], SIZE_Y-BORDER[1]-BORDER[3]);
      cs.set_pixel_start(BORDER[0],BORDER[1]);
      cs.set_value_ranges(x_min,0,x_max,y_max);
      //SwimlanesStateAction swimlaneslines = new SwimlanesStateAction(maingroup, cs);
      SwimlanesStateAction swimlaneslines = new SwimlanesStateAction(maingroup, cs);
      
      // add everything to the plot
      draw.add(xaxis);
      draw.add(yaxis);
      draw.add(swimlaneslines);
      
      AxisLabelLayout xlabels = new AxisLabelLayout("xlab", xaxis, xlabBound);
      this.viz.putAction("xlabels",xlabels);
      AxisLabelLayout ylabels = new AxisLabelLayout("ylab", yaxis, ylabBound);
      this.viz.putAction("ylabels",ylabels);
    }
    
    // add axes names
    {
      SpecifiedLayout sl = new SpecifiedLayout(labelgroup, VisualItem.X, VisualItem.Y);
      ActionList labeldraw = new ActionList();
      labeldraw.add(sl);
      this.viz.putAction(labelgroup, labeldraw);
    }
    
    // add legend
    if (plot_legend) {
      ShapeAction legend_sa = new ShapeAction(legendshapegroup);
      SpecifiedLayout legendlabels_sl = new SpecifiedLayout(legendgroup, VisualItem.X, VisualItem.Y);
    
      ActionList legenddraw = new ActionList();
      legenddraw.add(legend_sa);
      this.viz.putAction(legendshapegroup, legenddraw);
      ActionList legendlabelsdraw = new ActionList();
      legendlabelsdraw.add(legendlabels_sl);
      this.viz.putAction(legendgroup,legendlabelsdraw);
    }
    
    // draw everything else
    this.viz.putAction("draw",draw);

    // finally draw
    this.viz.run("draw");
    this.viz.run("xlabels");
    this.viz.run("ylabels");

  }
  
  @edu.umd.cs.findbugs.annotations.SuppressWarnings(value =
      "SQL_NONCONSTANT_STRING_PASSED_TO_EXECUTE", 
      justification = "Dynamic based upon tables in the database")
  public Table getData() {
    // preliminary setup
    OfflineTimeHandler time_offline;
    TimeHandler time_online;
    long start, end;
    
    if (offline_use) {
      time_offline = new OfflineTimeHandler(param_map, this.timezone);
      start = time_offline.getStartTime();
      end = time_offline.getEndTime();
    } else {
      time_online = new TimeHandler(this.request, this.timezone);
      start = time_online.getStartTime();
      end = time_online.getEndTime();
    }
    
    DatabaseWriter dbw = new DatabaseWriter(this.cluster);
    String query;
    
    // setup query
    if (this.shuffle_option != null && this.shuffle_option.equals("shuffles")) {
      query = "select job_id,friendly_id,start_time,finish_time,start_time_millis,finish_time_millis,status,state_name,hostname from ["+this.table+"] where finish_time between '[start]' and '[end]'";
    } else {
      query = "select job_id,friendly_id,start_time,finish_time,start_time_millis,finish_time_millis,status,state_name,hostname from ["+this.table+"] where finish_time between '[start]' and '[end]' and not state_name like 'shuffle_local' and not state_name like 'shuffle_remote'";
    }
    if (this.jobname != null) {
      query = query + " and job_id like '"+ this.jobname +"'";
    }
    Macro mp = new Macro(start,end,query);
    query = mp.toString() + " order by start_time";
    
    Table rs_tab = null;    
    DatabaseDataSource dds; 

    log.debug("Query: " + query);
    // execute query
    try {
      dds = ConnectionFactory.getDatabaseConnection(dbw.getConnection());
      rs_tab = dds.getData(query);
    } catch (prefuse.data.io.DataIOException e) {
      System.err.println("prefuse data IO error: " + e);
      log.warn("prefuse data IO error: " + e);
      return null;
    } catch (SQLException e) {
      System.err.println("Error in SQL: " + e + " in statement: " + query);
      log.warn("Error in SQL: " + e + " in statement: " + query);
      return null;
    }
    
    HashMap<String, Integer> state_counts = new HashMap<String, Integer>();
    for (int i = 0; i < rs_tab.getRowCount(); i++) {
      String curr_state = rs_tab.getString(i, "state_name");
      Integer cnt = state_counts.get(curr_state);
      if (cnt == null) {
        state_counts.put(curr_state, Integer.valueOf(1));
      } else {
        state_counts.remove(curr_state);
        state_counts.put(curr_state, Integer.valueOf(cnt.intValue()+1));
      }
    }
    
    log.info("Search complete: #cols: " + rs_tab.getColumnCount() + "; #rows: " + rs_tab.getRowCount());
    
    return rs_tab;
  }
  
}