/*
    * 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 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], new Integer(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;
     }
   }
 
   /**
    * 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 = new String("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 = new String("");
     this.timezone = new String("");
     this.shuffle_option = new String("");
     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 = new String(cluster);
     if (timezone != null) {
       this.timezone = new String(timezone);
     } else {
       this.timezone = null;
     }
     this.shuffle_option = new String(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 = new String(cluster);
     if (timezone != null) {
       this.timezone = new String(timezone);
     } else {
       this.timezone = null;
     }
     this.shuffle_option = new String(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 = new String(cluster);
     if (timezone != null) {
       this.timezone = new String(timezone);
     } else {
       this.timezone = null;
     }
     this.shuffle_option = new String(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 = new String(evt_type);
     } else {
       this.shuffle_option = new String("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 = new String(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",new String("Time/s"));
     textlabels_table.setString(0,"type",new String("xaxisname"));
     
     VisualTable textlabelsviz = this.viz.addTable(labelgroup, textlabels_table);
     textlabelsviz.setX(0,SIZE_X/2);
     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");
 
   }
   
   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>();
     HashSet<String> states = new HashSet<String>();
     for (int i = 0; i < rs_tab.getRowCount(); i++) {
       String curr_state = rs_tab.getString(i, "state_name");
       states.add(curr_state);
       Integer cnt = state_counts.get(curr_state);
       if (cnt == null) {
         state_counts.put(curr_state, new Integer(1));
       } else {
         state_counts.remove(curr_state);
         state_counts.put(curr_state, new Integer(cnt.intValue()+1));
       }
     }
     
     log.info("Search complete: #cols: " + rs_tab.getColumnCount() + "; #rows: " + rs_tab.getRowCount());
     
     return rs_tab;
   }
   
 }