/*
 * MiningMart Version 1.0
 * 
 * Copyright (C) 2006 Martin Scholz, Timm Euler, 
 *                    Daniel Hakenjos, Katharina Morik
 *
 * Contact: miningmart@ls8.cs.uni-dortmund.de
 *
 * A list of contributing developers (other than the copyright 
 * holders) can be found at
 * http://mmart.cs.uni-dortmund.de/downloads/download.html
 * 
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License
 * as published by the Free Software Foundation; either version 2
 * of the License, or (at your option) any later version.
 * 
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 * GNU General Public License for more details.
 * 
 * You should have received a copy of the GNU General Public License
 * along with this program, see the file MM_HOME/LICENSE; if not, write
 * to the Free Software Foundation, Inc., 51 Franklin Street, Fifth
 * Floor, Boston, MA 02110-1301, USA.
 */
package edu.udo.cs.miningmart.operator;

import java.sql.SQLException;
import java.util.Collection;
import java.util.Iterator;
import java.util.StringTokenizer;

import edu.udo.cs.miningmart.exception.M4CompilerError;
import edu.udo.cs.miningmart.exception.M4Exception;
import edu.udo.cs.miningmart.m4.BaseAttribute;
import edu.udo.cs.miningmart.m4.Column;
import edu.udo.cs.miningmart.m4.Columnset;
import edu.udo.cs.miningmart.m4.ConceptualDatatypes;
import edu.udo.cs.miningmart.m4.Feature;
import edu.udo.cs.miningmart.m4.RelationalDatatypes;
import edu.udo.cs.miningmart.m4.Value;
import edu.udo.cs.miningmart.m4.utils.Print;

/**
 * This operator denormalizes or flattens out certain attributes, which is called 
 * Pivotizing. This means that for each value that occurs in the column of 
 * one or two "index" attributes, a new attribute is created. This new attribute 
 * contains the aggregation over those values of the "pivot" attribute where 
 * the index attribute(s) took the value that corresponds to the new
 * attribute. 
 * 
 * Example where aggregation is done by summing up (I = index attribute, P = pivotised attribute):
 * 
 * Input              Output
 * 
 *  I | P          P_M | P_F
 * ------          ----------
 *  M | 4    =>     6  | 5
 *  M | 2           
 *  F | 5          
 *
 * @author Timm Euler
 * @version $Id: ReversePivotize.java,v 1.1 2006/05/19 16:24:05 euler Exp $
 */
public class ReversePivotize extends SingleCSOperator {

	// Parameter names for this Operator
	public static final String PARAMETER_PIVOT_ATTRS  = "ThePivotizedAttributes";
	public static final String PARAMETER_VALUES       = "IndexValues";
	public static final String PARAMETER_KEY_ATTR     = "TheKeyAttribute";
	public static final String PARAMETER_NAME_PIVOT   = "NameForPivotAttribute";	
	public static final String PARAMETER_NAMES_INDEX_ATTRS = "NamesForIndexAttributes";		
		
	// the select string for the columns that are taken over from the input:
	private String selectSimpleColumns = null;
	
	/**
	 * @see miningmart.compiler.operator.SingleCSOperator#getTypeOfNewColumnSet()
	 */
	public String getTypeOfNewColumnSet() {
		return Columnset.CS_TYPE_VIEW;
	}

	/**
	 * @see edu.udo.cs.miningmart.operator.SingleCSOperator#generateSQLDefinition(String)
	 */
	public String generateSQLDefinition(String selectPart) throws M4CompilerError {
		try {
			String sqlDef = "(SELECT " + selectPart + " FROM ";
			
			long stepId = this.getStep().getId();
			BaseAttribute[] oldPivotAttrs = this.getInputPivotAttrs();
			String[] indexValues = this.getIndexVals();
			String[] indexAttrNames = this.getNewIndexAttrs();
			String nameOfPivotAttribute = ((Value[]) this.getParameter(PARAMETER_NAME_PIVOT))[0].getValue();
			String selectInner = "";
			
			// create temporary views:
			for (int i = 0; i < oldPivotAttrs.length; i++) {
				
				// The parameters 'IndexValues' and 'ThePivotizedAttributes' are 
				// coordinated. A pivot attribute's values are assigned to 
				// one combination of index values. Here we only need to quote
				// the index values, then we can create virtual columns that take
				// these values.				
				
				String selectForView = this.selectSimpleColumns + ", " + 
							oldPivotAttrs[i].getName() + " AS " + nameOfPivotAttribute;
				selectInner = this.selectSimpleColumns + ", " + nameOfPivotAttribute;
				
				StringTokenizer st = new StringTokenizer(indexValues[i], ", ");
				if (st.countTokens() != indexAttrNames.length) {
					throw new M4CompilerError("Operator ReversePivotize: Each entry in parameter 'IndexValues' must have as many values as there are index attributes!");
				}
				int pos = 0;
				while (st.hasMoreTokens()) {
					selectForView += ", '" + st.nextToken().trim() + "' AS " + indexAttrNames[pos];
					selectInner += ", " + indexAttrNames[pos];
					pos++;
				}
				// create one view for each pivotised attribute:
				String tempViewDef = "CREATE OR REPLACE VIEW tmp_" + stepId + "_" + i +
							" AS (SELECT " + selectForView + " FROM (" +
							this.getInputConcept().getCurrentColumnSet().getCompleteSQLQuery() + "))";
				try {
					this.getM4Db().executeBusinessSqlWrite(tempViewDef);
					this.getM4Db().addViewToTrash(tempViewDef, this.getInputConcept().getCurrentColumnSet().getSchema(), stepId);
				}
				catch (SQLException sqle) {
					throw new M4CompilerError("Operator ReversePivotize: SQL error trying to create a temporary view.\n" +
							"Message: " + sqle.getMessage() + "\n" +
							"View definition was: " + tempViewDef + "\n");							
				}
			}
			
			// now take the union of all temporary views:
			for (int i = 0; i < oldPivotAttrs.length; i++) {
				sqlDef += "(SELECT " + selectPart + " FROM tmp_" + stepId + "_" + i + ") UNION ";
			}
			sqlDef = sqlDef.substring(0, sqlDef.length() - 7);
			sqlDef += ")";
			return sqlDef;
		}
		catch (M4Exception m4e) {
			throw new M4CompilerError("Operator Pivotize: M4Exception caught " +
			                          "when accessing input concept/its columnset: " + m4e.getMessage());
		}
	}

	/**
	 * This method is overridden because this operator creates a special output.
	 * 
	 * @see edu.udo.cs.miningmart.m4.operator.ConceptOperator#generateColumns(Columnset)
	 */
    protected String generateColumns(Columnset csForOutputConcept) throws M4CompilerError {
    	
		String columnExpr = ""; // to be returned
		this.selectSimpleColumns = ""; // to be used in method 'generateSQLDefinition'
		
		Feature[] theExistingPivotAttrs = this.getInputPivotAttrs();
		try	{
			// go through the Features of the output concept:
			Collection outFeatures = this.getOutputConcept().getFeatures();
			if (outFeatures == null || outFeatures.isEmpty()) {
				throw new M4CompilerError("Operator ReversePivotize: No Features found in the output concept!");
			}
			Iterator it = outFeatures.iterator();
		 L:	while (it.hasNext()) {
				Feature outF = (Feature) it.next();
				
				// for debugging, check that it's not one of the input pivot BAs:
				for (int i = 0; i < theExistingPivotAttrs.length; i++) {
					if (outF.correspondsTo(theExistingPivotAttrs[i])) {
						throw new M4CompilerError("Operator ReversePivotize: found an Input Pivot-BA in the Output Concept!");
					}
				}
				
				if (this.isNewIndexAttr(outF) || outF.getName().equalsIgnoreCase(this.getNewPivotAttr())) {
					// outF must be a newly created Attribute,
					// so its column is very simple 
					if ( ! (outF instanceof BaseAttribute)) {
						throw new M4CompilerError("ReversePivotize ('" + this.getStep().getName() +
								"'): cannot handle MultiColumnFeatures like '" +
								outF.getName() + "!");
					}
				}

	            int iIn = 0;
	            Feature inF = null;
                // find input feature with same name:
	            do {
		            inF = getInputConcept().getFeature(iIn);
		            iIn++;
                }
                while ((iIn < getInputConcept().getFeatures().size()) && ( ! outF.correspondsTo(inF)));

	            if (( ! outF.correspondsTo(inF)) && (this.isNewIndexAttr(outF) || outF.getName().equalsIgnoreCase(this.getNewPivotAttr()))) {
	            	// outF is an attribute that this operator creates;
	            	// so create column for it with right data type:
	            	String columnDatatype = null;
	            	if (outF.getName().equalsIgnoreCase(this.getNewPivotAttr())) {
	            		columnDatatype = this.getColDataTypeOfPivotAttr();
	            	}
	            	else {
	            		columnDatatype = this.getColDataTypeOfIndexAttr(outF.getName());
	            	}
	            	Column outCol = csForOutputConcept.createColumn(outF.getName(), columnDatatype);
	    		    this.getStep().addToTrash(outCol);
	    		    // link output column to output BA:
	    		    outCol.setBaseAttribute((BaseAttribute) outF);
	    		    // set SQL definition of output column
	    		    outCol.setSQLDefinition(outF.getName());
	    		    
	    			columnExpr += outF.getName() + ", ";            	
	            	
	            	continue;
	            }	            
	            
	            if ( ! outF.correspondsTo(inF)) {
	            	throw new M4CompilerError("Operator ReversePivotize ('" + this.getStep().getName() +
						"'): found Feature '" + outF.getName() + "' in output concept that should not be there!");
                }
                
                if (this.isDeselectedParameter(inF)) {
                    this.doPrint(Print.PARAM, "Output Concept '" + getOutputConcept().getName() + "': skipped feature '" +
                                  outF.getName() + "' because the corresponding input feature was deselected by " +
                                  "a FeatureSelection operator.");   	
                	continue;	
                }
                
                if ( ! this.getStep().isVisible(inF)) {
                	// skip input features that are not visible in this step!
                    continue;		                	
                }

                // copy metadata for column to output columnset                
            	Column outCol = ((BaseAttribute) inF).getCurrentColumn().copyColToCS(csForOutputConcept);
    		    this.getStep().addToTrash(outCol);
    		    // link output column to output BA:
    		    outCol.setBaseAttribute((BaseAttribute) outF);
    		    // set SQL definition of output column
    		    outCol.setSQLDefinition(outCol.getName());
    		    
    			columnExpr += outCol.getName() + ", ";          
                this.selectSimpleColumns += outCol.getName() + ", "; 
			} // end of loop through the output features
		}
		catch (M4Exception m4e) {
			throw new M4CompilerError("Operator ReversePivotize: M4 Exception caught when generating metadata: " +
			                          m4e.getMessage());
		}	
		if (columnExpr.length() > 2) {
			columnExpr = columnExpr.substring(0, columnExpr.length() - 2);
		}
		else {
			throw new M4CompilerError("Operator ReversePivotize: No columns created for output concept!");
		}
		if (this.selectSimpleColumns.length() > 2) {
			this.selectSimpleColumns = this.selectSimpleColumns.substring(0, this.selectSimpleColumns.length() - 2);
		}
		
		return columnExpr;		             
    }
    
	/**
	 * This method is never called.
	 * 
	 * @see miningmart.compiler.operator.ConceptOperator#mustCopyFeature(String)
	 */
	protected boolean mustCopyFeature(String nameOfFeature)	throws M4CompilerError {
		return false;
	}
	
	private String[] getNewIndexAttrs() throws M4CompilerError {
		Value[] vs = (Value[]) this.getParameter(PARAMETER_NAMES_INDEX_ATTRS);
		if (vs == null || vs.length == 0) {
			throw new M4CompilerError("Operator ReversePivotize: Parameter '" +
			                          PARAMETER_NAMES_INDEX_ATTRS + "' not found!");
		}
		String[] names = new String[vs.length]; 
		for (int i = 0; i < vs.length; i++) {
			names[i] = vs[i].getValue();
		}
		return names;
	}

	private String getColDataTypeOfPivotAttr() throws M4CompilerError {
		BaseAttribute[] pivBAs = this.getInputPivotAttrs();
		try {
			Column oneCol = pivBAs[0].getCurrentColumn();
			return oneCol.getColumnDataTypeName();
		}
		catch (M4Exception m4e) {
			throw new M4CompilerError("Operator ReversePivotize: M4 error accessing a column of an input Pivot attribute: " + m4e.getMessage());
		}
	}
	
	public static String getConDataTypeOfPivotAttr(BaseAttribute[] pivotAttrs) throws M4Exception {
		if (pivotAttrs != null && pivotAttrs.length > 0) {
			return pivotAttrs[0].getConceptualDataTypeName();
		}
		else return null;
	}
	
	public static String getConDataTypeOfIndexAttr(String nameOfIndexAttr, String[] indexValues, String[] indexAttrNames) 
	throws M4Exception {
		String[] values = ReversePivotize.getValuesOfIndexAttr(nameOfIndexAttr, indexValues, indexAttrNames);
		// check if there are only numbers:
		boolean allNumbers = true;
		for (int i = 0; i < values.length; i++) {
			try {
				Double.parseDouble(values[i]);
			}
			catch (NumberFormatException nfe) {
				allNumbers = false;
			}
		}
		if (allNumbers) {
			return ConceptualDatatypes.CDT_NUMERIC;
		}
		else return ConceptualDatatypes.CDT_NOMINAL;
	}
	
	private String getColDataTypeOfIndexAttr(String nameOfIndexAttr) 
	throws M4CompilerError {
		
		try {
			// we try to guess the column data type by inspecting the values
			// that this column takes.
			String[] indexVals = this.getIndexVals();
			String[] indexAttrNames = this.getNewIndexAttrs();
			
			String[] values = ReversePivotize.getValuesOfIndexAttr(nameOfIndexAttr, indexVals, indexAttrNames);
		// 	check if there are only numbers:
			boolean allNumbers = true;
			for (int i = 0; i < values.length; i++) {
				try {
					Double.parseDouble(values[i]);
				}
				catch (NumberFormatException nfe) {
					allNumbers = false;
				}
			}
			if (allNumbers) {
				return RelationalDatatypes.RELATIONAL_DATATYPE_NUMBER;			
			}
			else {
				return RelationalDatatypes.RELATIONAL_DATATYPE_STRING;
			}
		}
		catch (M4Exception m4e) {
			throw new M4CompilerError(m4e.getMessage());
		}
	}
	
	private static String[] getValuesOfIndexAttr(String nameOfIndexAttr, String[] indexVals, String[] indexAttrNames) 
	throws M4Exception {
		// the order of the index attributes determines which index values are assigned to
		// the given index attribute:
		int position = 0;
		while ( ! indexAttrNames[position].equalsIgnoreCase(nameOfIndexAttr)) {
			position++;
		}
		if ( ! indexAttrNames[position].equalsIgnoreCase(nameOfIndexAttr)) {
			throw new M4Exception("Operator ReversePivotize: Could not find output concept feature '" + nameOfIndexAttr + "' in list of index attributes!");			
		}
		// loop through all index values, each parameter gives a combination of values 
		// from each index attribute:
		String[] valuesOfGivenIndexAttr = new String[indexVals.length];
		for (int i = 0; i < indexVals.length; i++) {
			StringTokenizer st = new StringTokenizer(indexVals[i], ",");
			int posOfToken = -1;
			// collect all tokens at "position":
			String oneIndexValue = null;
			while (posOfToken < position && st.hasMoreTokens()) {
				oneIndexValue = st.nextToken().trim();
				posOfToken++;
			}
			if (posOfToken == position) {
				valuesOfGivenIndexAttr[i] = oneIndexValue;
			}
			else {
				throw new M4Exception("Operator ReversePivotize: List of index values '" + indexVals[i] + 
						              "' does not match the number of index attributes given!");
			}
		}
		return valuesOfGivenIndexAttr;
	}
	
	private BaseAttribute[] getInputPivotAttrs() throws M4CompilerError {
		BaseAttribute[] bas = (BaseAttribute[]) this.getParameter(PARAMETER_PIVOT_ATTRS);
		if (bas == null || bas.length == 0) {
			throw new M4CompilerError("Operator ReversePivotize: Parameter '" +
                    PARAMETER_PIVOT_ATTRS + "' not found!");
		}
		return bas;
	}
	
	private String getNewPivotAttr() throws M4CompilerError {
		Value[] vs = (Value[]) this.getParameter(PARAMETER_NAME_PIVOT);
		if (vs == null || vs.length == 0) {
			throw new M4CompilerError("Operator ReversePivotize: Parameter '" +
			                          PARAMETER_NAME_PIVOT + "' not found!");
		}
		return vs[0].getValue(); 
	}
	
	private boolean isNewIndexAttr(Feature outputConceptFeature) throws M4CompilerError {
		String[] namesOfNewIndexAttrs = this.getNewIndexAttrs();
		for (int i = 0; i < namesOfNewIndexAttrs.length; i++) {
			if (outputConceptFeature.getName().equalsIgnoreCase(namesOfNewIndexAttrs[i])) {
				return true;
			}
		}
		return false;
	}	
	
	private String[] getIndexVals() throws M4CompilerError {
		Value[] vals = (Value[]) this.getParameter(PARAMETER_VALUES);
		if (vals == null || vals.length == 0) {
			throw new M4CompilerError("Operator ReversePivotize: Parameter '" +
			                          PARAMETER_VALUES + "' not found!");
		}
		String[] st = new String[vals.length];
		for (int i = 0; i < vals.length; i++) {
			st[i] = vals[i].getValue();
		}
		return st;
	}
}
/*
 * Historie
 * --------
 * 
 * $Log: ReversePivotize.java,v $
 * Revision 1.1  2006/05/19 16:24:05  euler
 * New operator 'ReversePivotize'.
 *
 * Revision 1.5  2006/04/11 14:10:12  euler
 * Updated license text.
 *
 * Revision 1.4  2006/04/06 16:31:11  euler
 * Prepended license remark.
 *
 * Revision 1.3  2006/03/20 09:15:46  euler
 * Updated Pivotize to allow pivotisation without aggregation.
 *
 * Revision 1.2  2006/03/17 17:06:39  euler
 * *** empty log message ***
 *
 * Revision 1.1  2006/01/03 09:54:22  hakenjos
 * Initial version!
 *
 */
