/*
 * 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.Iterator;
import java.util.Vector;

import edu.udo.cs.miningmart.exception.DbConnectionClosed;
import edu.udo.cs.miningmart.exception.M4CompilerError;
import edu.udo.cs.miningmart.exception.M4Exception;
import edu.udo.cs.miningmart.exception.ParameterDeselectedError;
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.Concept;
import edu.udo.cs.miningmart.m4.Feature;
import edu.udo.cs.miningmart.m4.MultiColumnFeature;
import edu.udo.cs.miningmart.m4.RelationalDatatypes;
import edu.udo.cs.miningmart.m4.Step;
import edu.udo.cs.miningmart.m4.utils.Print;

/**
 * UnionByKey operator takes as an input concepts and selected features from
 * these conpcepts. Then it creates the output concept containing the data
 * found in all the input concepts by joining concepts' columnsets according 
 * to the specified key attributes.
 * It is useful in a situation that we would like to gather the data not
 * only for the common to all the input concepts subset of keys, as it is in
 * case of JoinByKey operator.
 *
 * The columnset generated for the output concept is of type table!
 * 
 * @author Timm Euler, Martin Scholz
 * @version $Id: UnionByKey.java,v 1.6 2006/04/11 14:10:11 euler Exp $
 */
public class UnionByKey extends SingleCSOperator {

	/** Caches the key of the output concept (see this.getTheOutputIdAttribute()) */
	private BaseAttribute theOutputId = null;

	/** @return parameter array &quot;TheConcepts&quot; */
	public Concept[] getTheConcepts() throws M4CompilerError {
		return (Concept[]) this.getParameter("TheConcepts");
	}

	/** @return parameter array &quot;TheKeys&quot; */
	public BaseAttribute[] getTheKeys() throws M4CompilerError {
		return (BaseAttribute[]) this.getParameter("TheKeys");
	}

	/**
	* Getter method for the mapping parameters.
	* 
	* @return A two-dimensional array of Features.
	*         At position [0][i] is the ith "MapInput",
	*         at position [1][i] is the ith "MapOutput".
	*/
	protected Feature[][] getInOutMap() throws M4CompilerError {
		Feature[] inputF = (Feature[]) this.getParameter("MapInput");
		Feature[] outputF = (Feature[]) this.getParameter("MapOutput");
		int noOfMappings = inputF.length;
		// this should have been checked already, but let's be safe:
		if (outputF.length != noOfMappings) {
			throw new M4CompilerError("Operator UnionByKey: need same number of MapInput and MapOutput attributes!");
		}
		final Feature[][] theMap = new Feature[2][noOfMappings];
		for (int l = 0; l < noOfMappings; l++) {
			theMap[0][l] = inputF[l];
			theMap[1][l] = outputF[l];
			if ((theMap[0][l] == null) || (theMap[1][l] == null)) {
				throw new M4CompilerError("Operator UnionByKey: mapping information is incomplete!");
			}
		}
		return theMap;
	}

	/**
	 * Tries to find the key of the output concept.
	 * Therefore it uses the defined mappings (parameter of the operator)
	 * and looks for corresponding attributes (by name) of the keys of the
	 * input concepts.
	 * @return the key of the output concept
	 * @throws M4CompilerError if no key can be identified
	 *  */
	public BaseAttribute getOutputIdAttribute() throws M4CompilerError {
		if (this.theOutputId != null) {
			return (this.theOutputId);
		}
		BaseAttribute[] keys = this.getTheKeys();
		for (int i = 0; i < keys.length; i++) {
			Feature f = this.getMappedFeature(keys[i]);
			if (f != null && f instanceof BaseAttribute) {
				this.theOutputId = (BaseAttribute) f;
				return this.theOutputId;
			}
		}
		throw new M4CompilerError("Operator UnionByKey: No Key found in output Concept!");
	}

	public void load(Step st) throws ParameterDeselectedError, M4CompilerError {
		super.load(st);
		checkKeys();
	}

	/**
	 * TheInputConcept now is the first concept of the loaded list.
	 */
	public Concept getInputConcept() throws M4CompilerError {
		return this.getTheConcepts()[0];
	}

	/**
	 * This method is to ensure that JoinAttribute[i] belongs to Concept[i]
	 */
	private void checkKeys() throws M4CompilerError {
		if (this.getTheKeys().length != this.getTheConcepts().length)
			throw new M4CompilerError("UnionByKey: number of JoinAttributes and number of Concepts must be equal!");
		boolean found;
		for (int i = 0; i < this.getTheKeys().length; i++) {
			Concept theConcept;
			try {
				theConcept = this.getTheKeys()[i].getConcept();
			}
	   		catch (M4Exception m4e)
	   		{   throw new M4CompilerError("M4 interface error in " + this.getName() + ": " + m4e.getMessage());  } 
			found = false;
			if (theConcept.getId() == this.getTheConcepts()[i].getId()) {
				found = true;
			}
			if (!found)
				throw new M4CompilerError(
					"Attribute "
						+ this.getTheKeys()[i].getName()
						+ " does not belong to the concept "
						+ this.getTheConcepts()[i].getName()
						+ " !");
		}
	}

	/**
	 * Always returns 'T' as the new columnset is a table
	 */
	public String getTypeOfNewColumnSet() {
		return Columnset.CS_TYPE_TABLE;
	}

	/**
	 * It is never called.
	 */
	protected boolean mustCopyFeature(String nameOfFeature) {
		return false;
	}

	/**
	 * SQL definition of the table created for the output concept is simply
	 * the name of this table.
	 */
	public String generateSQLDefinition(String selectPart) throws M4CompilerError {
		return getNewCSName();
	}

	/**
	 * Holds the information describing single concept and its attributes plus
	 * the key attribute.
	 */
	private class ConceptData {

		private Concept concept; // holds concept
		private Vector attributes; // holds selected BAs
		private BaseAttribute attribId; // holds the key BA
		private Iterator it = null;

		public ConceptData() {
			attributes = new Vector();
		}

		public Concept getConcept() {
			return concept;
		}

		public void setConcept(Concept cs) {
			this.concept = cs;
		}

		public BaseAttribute getIdAttribute() {
			return attribId;
		}

		public void setIdAttribute(BaseAttribute id) {
			this.attribId = id;
		}

		public void resetBA() {
			it = attributes.iterator();
		}

		public boolean hasNextBA() {
			if (it == null)
				it = attributes.iterator();
			return it.hasNext();
		}

		public BaseAttribute nextBA() {
			if (it.hasNext()) {
				return (BaseAttribute) it.next();
			}
			else {
				it = null;
			}
			return null;
		}

		public void addBA(BaseAttribute ba) {
			attributes.add(ba);
		}
	}

	/**
	 * Here information of all the input concepts is held which plays a role in the output
	 * concept creation.
	 */
	private Vector conceptsData;

	/**
	 * Converts type of a column to column type as it exists int the database.
	 * The database here is Oracle8i, so the types are NUMBER, DATE, VARCHAR2
	 */
	private String determineDBTypeOfColumn(Column col) throws M4CompilerError {
		boolean useBusinessDBMS = true;
		if (col.getColumnDataType() == 12)
			return this.getM4Db().getDbNameOfM4Datatype(RelationalDatatypes.RELATIONAL_DATATYPE_NUMBER, 0, useBusinessDBMS);
		if (col.getColumnDataType() == 13 || col.getColumnDataType() == 15)
			return this.getM4Db().getDbNameOfM4Datatype(RelationalDatatypes.RELATIONAL_DATATYPE_STRING, 500, useBusinessDBMS);// hmmm... 500 ?
		if (col.getColumnDataType() == 14)
			return this.getM4Db().getDbNameOfM4Datatype(RelationalDatatypes.RELATIONAL_DATATYPE_DATE, 0, useBusinessDBMS);
		// use string as default (??):
		return this.getM4Db().getDbNameOfM4Datatype(RelationalDatatypes.RELATIONAL_DATATYPE_STRING, 500, useBusinessDBMS);
	}
		
	/**
	 * If some of the features of the output concept do not have their counterpart
	 * in any of the input concepts just print warning message. This method is
	 * rather for debugging purposes and may be removed from the code.
	 */
	private void checkOutputConceptFeatures() throws M4CompilerError {
		try
		{
			Iterator outConFeas =
				this.getOutputConcept().getFeatures().iterator();
			Concept[] cons = this.getTheConcepts();
			boolean found;
			Feature outF;
			while (outConFeas.hasNext())
			{
				// OutputID attrib. does not have to be found in any of the input
				// // concepts
				outF = (Feature) outConFeas.next();
				if (outF.getId() == this.getOutputIdAttribute().getId())
					continue;
				found = false;
				for (int j = 0; j < cons.length; j++)
				{
					Iterator it = cons[j].getFeatures().iterator();
					while (it.hasNext())
					{
						if (outF.correspondsTo((Feature) it.next()))
							found = true;
					}
				}
				if (!found)
					this.doPrint(
						Print.OPERATOR,
						"UnionByKey: No Feature with id "
							+ outF.getId()
							+ " ("
							+ outF.getName()
							+ ") found in any of the input concepts. Skipping.");
			}
		}
   		catch (M4Exception m4e)
   		{   throw new M4CompilerError("M4 interface error in " + this.getName() + ": " + m4e.getMessage());  } 
	}

	
	/**
	 * Overrides the superclass method because the mapping is used.
	 * 
	 * @see miningmart.operator.ConceptOperator#generateColumns(Columnset)
	 */
	protected String generateColumns(Columnset csForOutputConcept)
		throws M4CompilerError {
		Column oldColumn = null;
		Column copiedColumn = null;
		/* changed on demand: it is not changed, but I want you to see this comment.
		 * I hope that the key attribute can not be deselected one.
		 */
		 try {
			oldColumn = this.getTheKeys()[0].getCurrentColumn();
			copiedColumn = oldColumn.copyColToCS(csForOutputConcept);
			this.getStep().addToTrash(copiedColumn);
			copiedColumn.setBaseAttribute(this.getOutputIdAttribute());
			// this.getOutputIdAttribute().addColumn(copiedColumn);
		 }
   		catch (M4Exception m4e)
   		{   throw new M4CompilerError("M4 interface error in " + this.getName() + ": " + m4e.getMessage());  } 
   		
		copiedColumn.setSQLDefinition(this.getOutputIdAttribute().getName());
		conceptsData = new Vector(); // data for all concepts
		for (int i = 0; i < this.getTheConcepts().length; i++) {
			generateConceptData(
				this.getTheConcepts()[i],
				this.getTheKeys()[i],
				csForOutputConcept);
		}
		this.workOnDb(csForOutputConcept);
		return "";
	}

	/* update 24.I.2003 */
	private boolean isAKey(Feature f) throws M4CompilerError {
		for (int i = 0; i < this.getTheKeys().length; i++) {
			if (f.getId() == this.getTheKeys()[i].getId()) return true;
		}
		return false;
	}
	/* end of update 24.I.2003 */

	
	/** 
	 * Processing of single concept and its selected attributes. Metadata
	 * creation and information collecting for further use.
	 */
	private void generateConceptData(
			Concept current,
			BaseAttribute idBA,
			Columnset csForOutputConcept)
		throws M4CompilerError 
	{
		try
		{
			ConceptData cData = new ConceptData();
			conceptsData.add(cData);
			// find selected features that belong to the current concept
			Vector selectedFeaturesOfCurrentConcept = new Vector();
			Iterator concFeatures = current.getFeatures().iterator();
			Feature currentFeature;
			L : while (concFeatures.hasNext())
			{

				currentFeature = (Feature) concFeatures.next();

				for (int j = 0;
					j < this.getOutputConcept().getFeatures().size();
					j++)
				{
					// additional check 21.02.2003:
					if (this.getMappedFeature(currentFeature) == null)
					{
						this.doPrint(
							Print.OPERATOR,
							"UnionByKey: Feature '"
								+ currentFeature.getName()
								+ "' not found in Mapping or in output concept; skipping!");
						continue L;
					}

					/* update 24.I.2003 */
					if (this
						.getMappedFeature(currentFeature)
						.correspondsTo(this.getOutputConcept().getFeature(j))
						&& // do not copy OutputID
					!this.isAKey(currentFeature))
						/* end of update 24.I.2003 */
					{
						// changed on demand: just do not add the deselected feature to the
						// ConceptData structure. I hope it will work, but I've got no chance 
						// to test it, because I'm not deselecting anything.
						if (!this.isDeselectedParameter(currentFeature))
							selectedFeaturesOfCurrentConcept.add(
								currentFeature);
					}
				}
			}
			// if no features must be selected from the current concept, skip that
			// concept

			/*  -- but keys are needed!?
			if (selectedFeaturesOfCurrentConcept.isEmpty()) {
				this.doPrint(
					Print.OPERATOR,
					"No features from concept '"
						+ current.getName()
						+ "' selected. Skipping.");
				return;
			}
			*/

			cData.setConcept(current);
			cData.setIdAttribute(idBA);
			Feature oldF, newF;
			BaseAttribute oldBA, newBA;
			MultiColumnFeature oldMCF, newMCF;
			Column copiedColumn, oldColumn;
			Iterator theBAs;
			// create new metadata for the selected features
			for (int i = 0; i < selectedFeaturesOfCurrentConcept.size(); i++)
			{
				oldF = (Feature) selectedFeaturesOfCurrentConcept.get(i);
				newF = this.getMappedFeature(oldF);
				// copy the metadata
				if (oldF instanceof BaseAttribute)
				{
					newBA = (BaseAttribute) newF;
					oldBA = (BaseAttribute) oldF;
					oldColumn = oldBA.getCurrentColumn();
					copiedColumn = oldColumn.copyColToCS(csForOutputConcept);
					this.getStep().addToTrash(copiedColumn);
					copiedColumn.setBaseAttribute(newBA);
					// newBA.addColumn(copiedColumn);
					copiedColumn.setName(newBA.getName());
					copiedColumn.setSQLDefinition(newBA.getName());
					cData.addBA(oldBA);
				}
				else
				{
					if (oldF instanceof MultiColumnFeature)
					{
						// MultiColumnFeature not supported yet:
						if (!newF.correspondsTo(oldF))
						{
							throw new M4CompilerError("Operator UnionByKey: Mapping of MultiColumnFeatures not implemented, yet!");
						}
						newMCF = (MultiColumnFeature) newF;
						oldMCF = (MultiColumnFeature) oldF;
						theBAs = newMCF.getBaseAttributes().iterator();
						if (theBAs != null)
						{
							while (theBAs.hasNext())
							{
								newBA = (BaseAttribute) theBAs.next();
								oldBA = oldMCF.getBaseAttributeByName(newBA.getName());
								oldColumn = oldBA.getCurrentColumn();
								copiedColumn = oldColumn.copyColToCS(csForOutputConcept);
								this.getStep().addToTrash(copiedColumn);
								copiedColumn.setBaseAttribute(newBA);
								// newBA.addColumn(copiedColumn);
								copiedColumn.setSQLDefinition(newBA.getName());
								cData.addBA(oldBA);
							}
						}
					}
					else
					{
						throw new M4CompilerError(
							"Unknown Feature type found in Concept with id: "
								+ current.getId()
								+ "; Feature id is "
								+ oldF.getId());
					}
				}
			}
		}
   		catch (M4Exception m4e)
   		{   throw new M4CompilerError("M4 interface error in " + this.getName() + ": " + m4e.getMessage());  } 
	}
			
	private String getMappedName(BaseAttribute ba) throws M4CompilerError {
		Feature outF = this.getMappedFeature(ba);
		if ((outF == null) || !(outF instanceof BaseAttribute)) {
			throw new M4CompilerError(
				"UnionByKey.getMappedName: No output BaseAttribute found for InputBA: "
					+ ba.getName()
					+ ", id: "
					+ ba.getId());
		}
		return outF.getName();
	}
	
	/** 
	 * Reads a Feature, finds the output Feature it is mapped to if a mapping exist.
	 * Otherwise the method looks for a corresponding Feature in the output concept
	 * by name.
	 * If neither a mapping nor a corresponding Feature in the input concepts exists
	 * <code>null</code> is returned
	 * @param an input Feature
	 * @return the output Feature or <code>null</code>
	 * */
	private Feature getMappedFeature(Feature oldF) throws M4CompilerError 
	{
		Feature[][] theMap = this.getInOutMap();
		for (int i = 0; i < theMap[0].length; i++) {
			Feature feature = theMap[0][i];
			if (feature.correspondsTo(oldF)) {
				return theMap[1][i];
			}
		}
		Feature newF = null;
		try {
			Iterator it = this.getOutputConcept().getFeatures().iterator();
			while (it.hasNext()
				   && 
				   (!((newF = (Feature) it.next()).correspondsTo(oldF)))
				  )
			{
				// just loop
			}
		}
   		catch (M4Exception m4e)
   		{   throw new M4CompilerError("M4 interface error in " + this.getName() + ": " + m4e.getMessage());  } 
   		
		// changed 21.02.2003:
		if (newF.correspondsTo(oldF))
		{   return newF;  }
		else
		{   return null;  }
	}




	// *********** Working on the business data ***********

	/** 
	 * This is the main method for all the work on the business database.
	 * @param csForOutputConcept the <code>Columnset</code> for the output
	 *        <code>Concept</code>
	 */
	private void workOnDb(Columnset csForOutputConcept)
		throws M4CompilerError
	{
		createTable(csForOutputConcept);
		if (this.conceptsData.size() == 2) {
			this.twoTableSolution(csForOutputConcept);
		}
		else {
			collectKeys(csForOutputConcept);
			fillAttributes(csForOutputConcept);
		}
	}

	/**
	 * This is the new method for UnionByKey, but up to now it works on
	 * two Concepts only! It excpects that the output table is already
	 * created, the meta-data structures of this operator are build and
	 * working, and it fills the output table.
	 */
	private void twoTableSolution(Columnset csForOutputConcept)
		throws M4CompilerError
	{

		// *** Gather all the necessary information first ***

		final String tmpTable = csForOutputConcept.getName() + "_TMP";
		final ConceptData first  = (ConceptData) this.conceptsData.get(0);
		final ConceptData second = (ConceptData) this.conceptsData.get(1);
		
		Column firstKey = null, secondKey = null;
		Columnset firstCS = null, secondCS = null;
		try {
			BaseAttribute id = first.getIdAttribute();
			if (id != null) {
				firstKey  = id.getCurrentColumn();
			}
			id = second.getIdAttribute();
			if (id != null) {
				secondKey = id.getCurrentColumn();
			}				
			Concept con = first.getConcept();
			if (con != null) {
				firstCS = con.getCurrentColumnSet();	
			}
			con = second.getConcept();
			if (con != null) {
				secondCS = con.getCurrentColumnSet();	
			}
			if (firstKey == null || secondKey == null || firstCS == null || secondCS == null) {
				throw new M4CompilerError(
					"UnionByKey: Found Key Attribute without Column or InputConcept without Columnset!");
			}	
		}
		catch (M4Exception e) {
			throw new M4CompilerError(e.getMessage());	
		}

		final String outTableName = csForOutputConcept.getSchemaPlusName();
		final String inTable1Name;
		final String inTable2Name;
		try {
			String viewPrefix = "TMP_" + this.getStep().getId() + "_";
			inTable1Name = this.createViewIfNecessary(firstCS, viewPrefix + "1");
			inTable2Name = this.createViewIfNecessary(secondCS, viewPrefix + "2");
		}
		catch (Exception e) {
			throw new M4CompilerError(
				"UnionByKey: Could not create a view for one of the input Columnset."
				+ "Exception message is:\n" + e.getMessage());
		}
		final String inTable1Key = this.curColNameForBa(first.getIdAttribute());
		final String inTable2Key = this.curColNameForBa(second.getIdAttribute());
		
		//final String outTableKey = this.curColNameForBa(this.getOutputIdAttribute());
		final String outTableKey = this.getOutputIdAttribute().getName();

		// *** create a temporary table with one attribute for keys ***
		this.createTmpKeyTable(tmpTable, firstKey);
		
		try {
			
			// *** Insert all keys part of both tables into the temporary table ***
			
			final String tmpKeyAttr = firstKey.getName();
			String query = "INSERT INTO " + tmpTable
						 + " ( " + tmpKeyAttr + " ) "
						 + " ( SELECT CS1."	+ inTable1Key
						 + " FROM " + inTable1Name + " CS1, "
						 + inTable2Name + " CS2"
						 + " WHERE CS1." + inTable1Key
						 + " = CS2." + inTable2Key + ")";
			try {
				this.executeBusinessSqlWrite(query);
			}
			catch (SQLException e) {
				throw new M4CompilerError(
					"SQLException when executing '" + query + "'\n:"
					+ e.getMessage());	
			}

			// *** Copy the disjoint tuples like this: ***

			// copy all tuples from table 1:			
			this.copyTuples(first, inTable1Name, outTableName);
			// remove all tuples with keys in both tables:
			this.removeIntersection(tmpTable, tmpKeyAttr, outTableName, outTableKey);
	
			// copy all tuples from table 2:
			this.copyTuples(second, inTable2Name, outTableName);
			// remove all tuples with keys in both tables:
			this.removeIntersection(tmpTable, tmpKeyAttr, outTableName, outTableKey);



			// ***** Finally insert the remaining tuples with a join *****

			try
			{		
				final String fromAttr, toAttr;
				final boolean secondTable;
				{
					String[] tga = this.getTargetAttributes(first, "CS1.");
					final String firstFromAttr = tga[0];
					final String firstToAttr = tga[1];
				
					tga = this.getTargetAttributes(second, "CS2.");
					String secondFromAttr = tga[0];
					String secondToAttr = tga[1];			
					// remove the id attribute which is already part of teh first attribute lists
			
					int firstComma = secondFromAttr.indexOf(',');
					if (firstComma == -1) {
						// No attributes from second table!
						fromAttr = firstFromAttr;	
						toAttr = firstToAttr;	
						secondTable = false;
					}
					else {
						secondFromAttr = secondFromAttr.substring(firstComma);
						secondToAttr = secondToAttr.substring(secondToAttr.indexOf(','));
					
						fromAttr = firstFromAttr + secondFromAttr;
						toAttr = firstToAttr + secondToAttr;
						secondTable = true;
					}
				}
								
				query = "INSERT INTO " + outTableName + " ( " + toAttr + " ) "
				      + "( SELECT " + fromAttr + " FROM " + inTable1Name + " CS1, "
				      + (secondTable ? (inTable2Name + " CS2, ") : "")
				      + tmpTable + " T"
				      + " WHERE CS1." + inTable1Key + " = T." + tmpKeyAttr
				      + (secondTable ? (" AND CS2." + inTable2Key   + " = T." + tmpKeyAttr) : "")
			    	  + " )";
				      
				this.executeBusinessSqlWrite(query);
			}
			catch (M4Exception e) {
				throw new M4CompilerError(
					"UnionByKey: Could not get attribute lists from ConceptData!\n"
					+ e.getMessage());
			}
			catch (SQLException e) {
				throw new M4CompilerError(
					"UnionByKey: Error executing write statement\n'"
					+ query + "'\non business data!");			
			}
		}
		finally {
			// Try to delete the temporary table in any case:
			try {
				this.getM4Db().dropBusinessTable(tmpTable);
			}
			catch (Exception e) {
				this.doPrint(Print.MAX,
					"UnionByKey: Tryied to remove '" + tmpTable + "' from business schema!");
				e.printStackTrace();
			}				
		}
	}

	/**
	 * This method checks for virtual attributes in the Columnset.
	 * If no such column is found then nothing is done at database level and
	 * the name of the Columnset provided is returned.
	 * Otherwise a view with the name specified is created in the business
	 * database and its name is returned. This view is also added to the 
	 * garbage collection, so the calling method does not have to care about
	 * this.
	 */
	private String createViewIfNecessary(Columnset inputCs, String viewName)
		throws M4Exception, DbConnectionClosed, SQLException, M4CompilerError
	{
		Iterator it = inputCs.getColumns().iterator();
		StringBuffer sbuf = new StringBuffer();

		boolean necessary = false;
	 	while (it.hasNext()) {
			Column cur = (Column) it.next();
			String sqlDef = cur.getSQLDefinition();
			if ( sqlDef != null && sqlDef.length() > 0
				&& ! sqlDef.equals(cur.getName()))
			{
				sbuf.append("( " + sqlDef + " ) AS " + cur.getName() + ", ");
				necessary = true;
			}
			else sbuf.append(cur.getName() + ", ");
		}
		if (necessary) {
			String viewSql = "SELECT " + sbuf.substring(0, sbuf.length() - 2)
			               + " FROM " + inputCs.getSchemaPlusName();

			String create = "CREATE OR REPLACE VIEW " + viewName + " AS ( " + viewSql + " )";
			
			this.getM4Db().executeBusinessSqlWrite(create);
			
			String schema = inputCs.getSchema();
			this.getM4Db().addViewToTrash(viewName, schema, this.getStep().getId());
			return schema + "." + viewName;
		}
		else return inputCs.getSchemaPlusName();
	}

	/** Helper method returning the name of the current column for a base attribute. */
	private String curColNameForBa(BaseAttribute ba) throws M4CompilerError {
		if (ba == null)
			return null;
			
		edu.udo.cs.miningmart.m4.Column column = null;
		try {
			column = ba.getCurrentColumn();	
		}
		catch (M4Exception e) {
			throw new M4CompilerError(
				"UnionByKey: Could not get current column of BaseAttribute '"
				+ ba.getName() + "'!\n" + e.getMessage());
		}
		
		return (column == null ? null : column.getName());
	}

	/**
	 * Copies the data of a complete <code>Columnset</code> to the output
	 * table.
	 * @param cd the <code>ConceptData</code> object of the input <code>Concept</code>
	 *        under consideration
	 * @param inTableName the name of the input table under consideration
	 * @param outTableName the name of the target table
	 */
	private void copyTuples(ConceptData cd, String inTableName, String outTableName)
		throws M4CompilerError
	{
		String[] tga;
		try {
			tga = this.getTargetAttributes(cd, null);
		}
		catch (M4Exception e) {
			throw new M4CompilerError("UnionByKey:\n" + e.getMessage());
		}
		String fromAttr = tga[0];
		String toAttr   = tga[1];
	
		String innerQ = "SELECT " + fromAttr + " FROM " + inTableName;

		String query = "INSERT INTO " + outTableName
	 			     + " ( " + toAttr + " ) ( " + innerQ + " )";

		try {
			this.executeBusinessSqlWrite(query);		
		}
		catch (SQLException e) {
			throw new M4CompilerError(
				"UnionByKey: Error executing write statement\n'"
				+ query + "'\non business data!");
		}
	}

	/**
	 * After inserting each of the two tables completely the tuples with
	 * keys part of both tables (thus stored in the temporary table) are
	 * deleted again by this method.
	 * @param tmpTable name of the temporary table in the business schema
	 * @param tmpKeyName the name of the attribute of the temporary table
	 * @param tableOut the target table from which we delete the tuples
	 * @param outKeyName the name of the key attribute in the target table
	 */
	private void removeIntersection(String tmpTable, String tmpKeyName, String tableOut, String outKeyName)
		throws M4CompilerError
	{
		String query = "DELETE FROM " + tableOut + " WHERE " + outKeyName
					 + " IN ( SELECT " + tmpKeyName + " FROM " + tmpTable + " )";
		try {
			// we need a commit to avoid Postgres messing up its transaction blocks...
			// though it's hard to know why...
			this.getM4Db().commitBusinessTransactions();
			this.executeBusinessSqlWrite(query);
		}
		catch (SQLException e) {
			throw new M4CompilerError(
				"UnionByKey: SQLException when writing\n'" + query
				+ "'\nto the business data:\n" + e.getMessage());
		}
	}
		
	/**
	 * Creates a temporary table in the business data schema to hold the keys
	 * contained in both tables. If a table with that name exists it is deleted!
	 * Please also note, that the table is <b>not</b> put on the garbage collection
	 * list, because it is deleted already after normal operator execution.
	 * 
	 * @param tableName the name of the new table
	 * @param key the <code>Column</code> of the first input columnset.
	 *        It is used to get the name and datatype of the (only) attribute
	 *        of the new table.
	 */
	private void createTmpKeyTable(String tableName, Column key)
		throws M4CompilerError
	{
		final String idxName = tableName + "_PK";
		
		try {
			this.getM4Db().dropBusinessTable(tableName);
		}
		catch (M4Exception m4e) {
			throw new M4CompilerError(m4e.getMessage());
		}
		
		String cQuery =
			"CREATE TABLE " + tableName + " ( " + key.getName() + " "
			+ determineDBTypeOfColumn(key) + ", CONSTRAINT " + idxName
			+ " PRIMARY KEY ( " + key.getName() + " ))";

		try {
			this.executeBusinessSqlWrite(cQuery);
		}
		catch (SQLException sqle) {
			throw new M4CompilerError("UnionByKey: SQL error: " + sqle.getMessage());
		}
	}

	/**
	 * @param cd the <code>ConceptData</code> object for the input <code>Concept</code>
	 *        under consideration
	 * @param tablePrefix a <code>String </code> like &quot;TableName.&quot; to place
	 *        in front of the attribute names of the input list to avoid ambigious
	 *        references. May be <code>null</code> if not needed.
	 * 
	 * @return two Strings (thus a <code>String[2]</code> object) which hold comma
	 *         separated attribute lists.
	 *         The first String is a list of all the input attributes of the Concept
	 * 		   for which the ConceptData is provided.
	 * 		   The second list is the list of attributes in the output table.
	 *         The mapping between these lists is given by the order of attributes.
	 *         The key attribute is the first one in both attribute lists.
	 */
	private String[] getTargetAttributes(ConceptData cd, String tablePrefix)
		throws M4CompilerError, M4Exception
	{
		if (tablePrefix == null || tablePrefix.trim().length() == 0) {
			tablePrefix = "";
		}
		else {
			tablePrefix = tablePrefix.trim();
			if (tablePrefix.charAt(tablePrefix.length() - 1) != '.')
				tablePrefix += ".";
		}
		
		StringBuffer sbufFrom = new StringBuffer();
		StringBuffer sbufTo = new StringBuffer();

		String idNameFrom = cd.getIdAttribute().getCurrentColumn().getName();
		sbufFrom.append(tablePrefix + idNameFrom + ", ");
		
		BaseAttribute mappedId;
		{
			Feature f = this.getMappedFeature(cd.getIdAttribute());
			if (f == null || ! (f instanceof BaseAttribute)) {
				throw new M4CompilerError("UnionByKey: No target BaseAttribute key found for input key!");	
			}
			mappedId = (BaseAttribute) f;
		}
		// old:
		// String idNameTo = mappedId.getCurrentColumn().getName();
		// new:
		String idNameTo = mappedId.getName();
		
		sbufTo.append(idNameTo + ", ");
		
		cd.resetBA();
		while (cd.hasNextBA()) {
			BaseAttribute ba = cd.nextBA();
			sbufFrom.append(tablePrefix + ba.getCurrentColumn().getName() + ", ");

			Feature outF = this.getMappedFeature(ba);
			if (outF != null && outF instanceof BaseAttribute) { // ignore MCF for now
				BaseAttribute outBa = (BaseAttribute) outF;
				if (outBa.getCurrentColumn() != null)
				    // old:
					// sbufTo.append(outBa.getCurrentColumn().getName() + ", ");
					// new:
					sbufTo.append(outBa.getName() + ", ");
			}
		}
		String[] ret = new String[2];
		ret[0] = sbufFrom.substring(0, sbufFrom.length() - 2);
		ret[1] = sbufTo.substring(0, sbufTo.length() - 2);
		return ret;
	}

	/** 
	 * Method for creating the table for output columnset
	 */
	private void createTable(Columnset cs) throws M4CompilerError {
		final String tableName = cs.getName();
		final String schema = cs.getSchema();
		final String idxName = tableName + "_PK";
		final long stepId = this.getStep().getId();
		
		try {
			this.getM4Db().dropBusinessTable(tableName);
		}
		catch (M4Exception m4e) {
			throw new M4CompilerError(m4e.getMessage());
		}
		
		String cQuery = "CREATE TABLE " + tableName + " (";
		try {
			final String idOutAttribute = this.getOutputIdAttribute().getName();
			cQuery += idOutAttribute + " "
			       + determineDBTypeOfColumn(this.getTheKeys()[0].getCurrentColumn())
			       + ", ";

			// loop over all the concepts
			for (int i = 0; i < conceptsData.size(); i++) {
				ConceptData cd = (ConceptData) conceptsData.get(i);
				cd.resetBA();
				while (cd.hasNextBA()) {
					BaseAttribute attr = cd.nextBA();
					cQuery += this.getMappedName(attr)
						   + " " + determineDBTypeOfColumn(attr.getCurrentColumn())
						   + ", ";
				}
			}
			cQuery += " CONSTRAINT " + idxName
			       + " PRIMARY KEY ( " + idOutAttribute + " ))";
			try {	
				this.executeBusinessSqlWrite(cQuery);
				this.getM4Db().addTableToTrash(tableName, schema, stepId);
			}
			catch (SQLException sqle) {
				throw new M4CompilerError("UnionByKey: SQL error: " + sqle.getMessage());
			}
		}
   		catch (M4Exception m4e)
   		{   throw new M4CompilerError("M4 interface error in " + this.getName() + ": " + m4e.getMessage());  } 
	}



	// *** These two method still do the work if there are not exactly two Concepts given ! ***
	
	/**
	 * This method collects all the keys found in the input concepts. Then id
	 * attribute for the output concept is created containing all possible key
	 * values.
	 */
	private void collectKeys(Columnset cs) throws M4CompilerError {
		try {
			String iQuery =
				"INSERT INTO "
					+ cs.getName()
					+ " ("
					+ this.getOutputIdAttribute().getName()
					+ ") (SELECT ID FROM (";
			for (int i = 0; i < conceptsData.size(); i++) {
				ConceptData cd = (ConceptData) conceptsData.get(i);
				iQuery += "(SELECT "
					+ cd.getIdAttribute().getCurrentColumn().getSQLDefinition()
					+ " AS ID FROM "
					+ cd.getConcept().getCurrentColumnSet().getSQLDefinition()
					+ ") UNION ";
			}
			iQuery = iQuery.substring(0, iQuery.length() - 7) + "))";
		
			this.getM4Db().executeBusinessSqlWrite(iQuery);
		}
		catch (SQLException sqle) {
			throw new M4CompilerError(
				"UnionByKey: SQL error occured: " + sqle.getMessage());
		}
   		catch (M4Exception m4e)
   		{   throw new M4CompilerError("M4 interface error in " + this.getName() + ": " + m4e.getMessage());  } 
	}
	
	/**
	 * Copying the data of selected attributes from input concepts to the output concept.
	 */
	private void fillAttributes(Columnset cs) throws M4CompilerError {
		String commonPart = "UPDATE " + cs.getName() + " CS " + " SET ";
		try {
			for (int i = 0; i < conceptsData.size(); i++) {
				ConceptData cd = (ConceptData) conceptsData.get(i);
				// Changed by NIT: Cannot use getSQLDefinition(). 
				// Altered to getName().
				/*String csSql =
					cd.getConcept().getCurrentColumnSet().getSQLDefinition();*/
				String csSql = cd.getConcept().getCurrentColumnSet().getName();
				cd.resetBA();
				while (cd.hasNextBA()) {
					BaseAttribute ba = cd.nextBA();
					String iQuery =
						this.getMappedName(ba)
							+ " = (SELECT "
							+ ba.getCurrentColumn().getSQLDefinition()
							+ " FROM "
							+ csSql
							+ " WHERE CS."
							+ this.getOutputIdAttribute().getName()
							+ " = "
							+ cd
								.getIdAttribute()
								.getCurrentColumn()
								.getSQLDefinition()
							+ ")";
					this.getM4Db().executeBusinessSqlWrite(commonPart + iQuery);
				}
			}
			this.getM4Db().commitBusinessTransactions();
		}
		catch (SQLException sqle) {
			throw new M4CompilerError(
				"UnionByKey: SQL exception occured: " + sqle.getMessage());
		}
   		catch (M4Exception m4e)
   		{   throw new M4CompilerError("M4 interface error in " + this.getName() + ": " + m4e.getMessage());  } 
	}
}
/*
 * Historie
 * --------
 *
 * $Log: UnionByKey.java,v $
 * Revision 1.6  2006/04/11 14:10:11  euler
 * Updated license text.
 *
 * Revision 1.5  2006/04/06 16:31:11  euler
 * Prepended license remark.
 *
 * Revision 1.4  2006/03/30 16:07:13  scholz
 * fixed author tags for release
 *
 * Revision 1.3  2006/03/29 09:50:47  euler
 * Added installation robustness.
 *
 * Revision 1.2  2006/03/23 11:13:45  euler
 * Improved exception handling.
 *
 * Revision 1.1  2006/01/03 09:54:22  hakenjos
 * Initial version!
 *
 */
