gt:             insertRowData = new Object[data.valuenrs.length];
context:  /* Copyright (c) 2007 Pentaho Corporation.  All rights reserved. 
   * This software was developed by Pentaho Corporation and is provided under the terms 
   * of the GNU Lesser General Public License, Version 2.1. You may not use 
   * this file except in compliance with the license. If you need a copy of the license, 
   * please go to http://www.gnu.org/licenses/lgpl-2.1.txt. The Original Code is Pentaho 
   * Data Integration.  The Initial Developer is Pentaho Corporation.
   *
   * Software distributed under the GNU Lesser Public License is distributed on an "AS IS" 
   * basis, WITHOUT WARRANTY OF ANY KIND, either express or  implied. Please refer to 
   * the license for the specific language governing your rights and limitations.*/
   
  package com.panet.imeta.trans.steps.tableoutput;
  
  import java.sql.BatchUpdateException;
  import java.sql.PreparedStatement;
  import java.sql.SQLException;
  import java.text.SimpleDateFormat;
  import java.util.ArrayList;
  import java.util.Date;
  import java.util.List;
  
  import com.panet.imeta.core.Const;
  import com.panet.imeta.core.RowMetaAndData;
  import com.panet.imeta.core.database.Database;
  import com.panet.imeta.core.database.DatabaseMeta;
  import com.panet.imeta.core.exception.KettleDatabaseBatchException;
  import com.panet.imeta.core.exception.KettleDatabaseException;
  import com.panet.imeta.core.exception.KettleException;
  import com.panet.imeta.core.exception.KettleStepException;
  import com.panet.imeta.core.row.RowDataUtil;
  import com.panet.imeta.core.row.RowMeta;
  import com.panet.imeta.core.row.RowMetaInterface;
  import com.panet.imeta.core.row.ValueMetaInterface;
  import com.panet.imeta.trans.Trans;
  import com.panet.imeta.trans.TransMeta;
  import com.panet.imeta.trans.step.BaseStep;
  import com.panet.imeta.trans.step.StepDataInterface;
  import com.panet.imeta.trans.step.StepInterface;
  import com.panet.imeta.trans.step.StepMeta;
  import com.panet.imeta.trans.step.StepMetaInterface;
  
  /**
   * Writes rows to a database table.
   * 
   * @author Matt Casters
   * @since 6-apr-2003
   */
  public class TableOutput extends BaseStep implements StepInterface
  {
  	private TableOutputMeta meta;
  	private TableOutputData data;
  		
  	public TableOutput(StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta, Trans trans)
  	{
  		super(stepMeta, stepDataInterface, copyNr, transMeta, trans);
  	}
  	
  	public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException
  	{
  		meta=(TableOutputMeta)smi;
  		data=(TableOutputData)sdi;
  
  		Object[] r=getRow();    // this also waits for a previous step to be finished.
  		if (r==null)  // no more input to be expected...
  		{
  			return false;
  		}
          
          if (first)
          {
              first=false;
              data.outputRowMeta = getInputRowMeta().clone();
              meta.getFields(data.outputRowMeta, getStepname(), null, null, this);
                
              if ( ! meta.specifyFields() )  {
              	// Just take the input row
              	data.insertRowMeta = getInputRowMeta().clone();
              }
              else  {
              	
              	data.insertRowMeta = new RowMeta();
              	
              	// 
              	// Cache the position of the compare fields in Row row
              	//
              	data.valuenrs = new int[meta.getFieldDatabase().length];
              	for (int i=0;i<meta.getFieldDatabase().length;i++)
              	{
              		data.valuenrs[i]=getInputRowMeta().indexOfValue(meta.getFieldStream()[i]);
              		if (data.valuenrs[i]<0)
              		{
              			throw new KettleStepException(Messages.getString("TableOutput.Exception.FieldRequired",meta.getFieldStream()[i])); //$NON-NLS-1$
              		}
              	}
  
          	    for (int i=0;i<meta.getFieldDatabase().length;i++) 
          	    {
          	 	    ValueMetaInterface insValue = getInputRowMeta().searchValueMeta( meta.getFieldStream()[i]); 
          		    if ( insValue != null )
          		    {
          			    ValueMetaInterface insertValue = insValue.clone();
          			    insertValue.setName(meta.getFieldDatabase()[i]);
          			    data.insertRowMeta.addValueMeta( insertValue );
          		    }
          		    else  {
          			    throw new KettleStepException(Messages.getString("TableOutput.Exception.FailedToFindField", meta.getFieldStream()[i])); //$NON-NLS-1$ 
          			}
          	    }            	
              }
          }
          
  		try
  		{
  			Object[] outputRowData = writeToTable(getInputRowMeta(), r);
              if (outputRowData!=null)
              {
                  putRow(data.outputRowMeta, outputRowData); // in case we want it go further...
                  incrementLinesOutput();
              }
              
              if (checkFeedback(getLinesRead())) 
              {
              	if(log.isBasic()) logBasic("linenr "+getLinesRead()); //$NON-NLS-1$
              }
  		}
  		catch(KettleException e)
  		{
  			logError("Because of an error, this step can't continue: ", e);
  			setErrors(1);
  			stopAll();
  			setOutputDone();  // signal end to receiver(s)
  			return false;
  		}		
  		
  		return true;
  	}
  
  	private Object[] writeToTable(RowMetaInterface rowMeta, Object[] r) throws KettleException
  	{
          
  		if (r==null) // Stop: last line or error encountered 
  		{
  			if (log.isDetailed()) logDetailed("Last line inserted: stop");
  			return null;
  		}
  
          PreparedStatement insertStatement = null;
          Object[] insertRowData; 
          Object[] outputRowData = r;
          
          String tableName = null;
          
          boolean sendToErrorRow=false;
          String errorMessage = null;
          boolean rowIsSafe = false;
          int[] updateCounts = null;
          List<Exception> exceptionsList = null;
          boolean batchProblem = false;
          Object generatedKey = null;
          
          if ( meta.isTableNameInField() )
          {
              // Cache the position of the table name field
              if (data.indexOfTableNameField<0)
              {
              	String realTablename=environmentSubstitute(meta.getTableNameField());
                  data.indexOfTableNameField = rowMeta.indexOfValue(realTablename);
                  if (data.indexOfTableNameField<0)
                  {
                      String message = "Unable to find table name field ["+realTablename+"] in input row";
                      log.logError(toString(), message);
                      throw new KettleStepException(message);
                  }
                  if (!meta.isTableNameInTable() && !meta.specifyFields())
                  {
                      data.insertRowMeta.removeValueMeta(data.indexOfTableNameField);
                  }
              }
              tableName = rowMeta.getString(r, data.indexOfTableNameField);
              if ( !meta.isTableNameInTable() && !meta.specifyFields() ) {
              	// If the name of the table should not be inserted itself, remove the table name
              	// from the input row data as well.  This forcibly creates a copy of r
              	//
              	insertRowData = RowDataUtil.removeItem(rowMeta.cloneRow(r), data.indexOfTableNameField);
              }
              else 
              {
                  insertRowData = r;
              }
          }
          else
          if (  meta.isPartitioningEnabled() && 
              ( meta.isPartitioningDaily() || meta.isPartitioningMonthly()) &&
              ( meta.getPartitioningField()!=null && meta.getPartitioningField().length()>0 )
             )
          {
              // Initialize some stuff!
              if (data.indexOfPartitioningField<0)
              {
                  data.indexOfPartitioningField = rowMeta.indexOfValue(environmentSubstitute(meta.getPartitioningField()));
                  if (data.indexOfPartitioningField<0)
                  {
                      throw new KettleStepException("Unable to find field ["+meta.getPartitioningField()+"] in the input row!");
                  }
  
                  if (meta.isPartitioningDaily())
                  {
                      data.dateFormater = new SimpleDateFormat("yyyyMMdd");
                  }
                  else
                  {
                      data.dateFormater = new SimpleDateFormat("yyyyMM");
                  }
              }
              
              ValueMetaInterface partitioningValue = rowMeta.getValueMeta(data.indexOfPartitioningField);
              if (!partitioningValue.isDate() || r[data.indexOfPartitioningField]==null)
              {
                  throw new KettleStepException("Sorry, the partitioning field needs to contain a data value and can't be empty!");
              }
              
              Object partitioningValueData = rowMeta.getDate(r, data.indexOfPartitioningField); 
              tableName=environmentSubstitute(meta.getTablename())+"_"+data.dateFormater.format((Date)partitioningValueData);
              insertRowData = r;
          }
          else
          {
              tableName  = data.tableName;
              insertRowData = r;
          }
          
          if ( meta.specifyFields() )  {
          	//
  			// The values to insert are those in the fields sections
              //
  