# HG changeset patch
# User Max Hofer <mh182@gmx.net>
# Date 1313005127 -7200
# Branch trunk
# Node ID 3402ecb6225ccb607dda19e92fb1da12a867c792
# Parent  35f3652aa7b3fae4f2f82875adf42fec807d162c
Refactor InsertUpdate.processRow (first) - REVIEW NEEDED

Split functionality in simpler, smaller methods. Move those methods to
InsertUpdateMeta (code smell: InsertUpdate called a lot of methods loacted in
InsertUpdateMeta class). The InsertUpdateMeta class is used to fill
InsertUpdateData.

NOTE: review needed - some logging statements were disabled since we have no
access to the logging object in InsertUpdateMeta. Is there a way to log from the
InsertUpdateMeta class?

diff -r 35f3652aa7b3 -r 3402ecb6225c src/org/pentaho/di/trans/steps/insertupdate/InsertUpdate.java
--- a/src/org/pentaho/di/trans/steps/insertupdate/InsertUpdate.java	Wed Aug 10 21:38:44 2011 +0200
+++ b/src/org/pentaho/di/trans/steps/insertupdate/InsertUpdate.java	Wed Aug 10 21:38:47 2011 +0200
@@ -11,18 +11,11 @@
  
 package org.pentaho.di.trans.steps.insertupdate;
 
-import java.sql.SQLException;
-import java.util.ArrayList;
-import java.util.List;
 
-import org.apache.commons.lang.ArrayUtils;
-import org.pentaho.di.core.Const;
 import org.pentaho.di.core.database.Database;
-import org.pentaho.di.core.database.DatabaseMeta;
 import org.pentaho.di.core.exception.KettleDatabaseException;
 import org.pentaho.di.core.exception.KettleException;
 import org.pentaho.di.core.exception.KettleStepException;
-import org.pentaho.di.core.row.RowMeta;
 import org.pentaho.di.core.row.RowMetaInterface;
 import org.pentaho.di.core.row.ValueMetaInterface;
 import org.pentaho.di.i18n.BaseMessages;
@@ -45,10 +38,10 @@
  */
 public class InsertUpdate extends BaseStep implements StepInterface
 {
-	private static Class<?> PKG = InsertUpdateMeta.class; // for i18n purposes, needed by Translator2!!   $NON-NLS-1$
+    private static Class<?> PKG = InsertUpdateMeta.class; // for i18n purposes, needed by Translator2!! $NON-NLS-1$
 
-	private InsertUpdateMeta meta;
-	private InsertUpdateData data;
+    private InsertUpdateMeta meta;
+    private InsertUpdateData data;
 	
 	public InsertUpdate(StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta, Trans trans)
 	{
@@ -189,103 +182,11 @@
         
         if (first)
         {
-            first=false;
-            
-            data.outputRowMeta = getInputRowMeta().clone();
-            meta.getFields(data.outputRowMeta, getStepname(), null, null, this);
-            
-            data.schemaTable = meta.getDatabaseMeta().getQuotedSchemaTableCombination(environmentSubstitute(meta.getSchemaName()), 
-            		                                                                  environmentSubstitute(meta.getTableName()));
-            
-            // lookup the values!
-            if (log.isDebug()) logDebug(BaseMessages.getString(PKG, "InsertUpdate.Log.CheckingRow")+getInputRowMeta().getString(r)); //$NON-NLS-1$
-
-            ArrayList<Integer> keynrs = new ArrayList<Integer>(meta.getKeyStream().length);
-            ArrayList<Integer> keynrs2 = new ArrayList<Integer>(meta.getKeyStream().length);
-            
-            for (int i=0;i<meta.getKeyStream().length;i++)
-            {
-            	int keynr = getInputRowMeta().indexOfValue(meta.getKeyStream()[i]);
-            	
-                if (keynr<0 &&  // couldn't find field!
-                    !"IS NULL".equalsIgnoreCase(meta.getKeyCondition()[i]) &&   // No field needed! //$NON-NLS-1$
-                    !"IS NOT NULL".equalsIgnoreCase(meta.getKeyCondition()[i])  // No field needed! //$NON-NLS-1$
-                   )
-                {
-                    throw new KettleStepException(BaseMessages.getString(PKG, "InsertUpdate.Exception.FieldRequired",meta.getKeyStream()[i])); //$NON-NLS-1$ //$NON-NLS-2$
-                }
-                keynrs.add(keynr);
-                
-                // this operator needs two bindings
-                if ("= ~NULL".equalsIgnoreCase(meta.getKeyCondition()[i])){
-                	keynrs.add(keynr);
-                	keynrs2.add(-1);
-                }
-                
-                int keynr2 = getInputRowMeta().indexOfValue(meta.getKeyStream2()[i]);
-                if (keynr2<0 &&  // couldn't find field!
-                    "BETWEEN".equalsIgnoreCase(meta.getKeyCondition()[i])   // 2 fields needed! //$NON-NLS-1$
-                   )
-                {
-                    throw new KettleStepException(BaseMessages.getString(PKG, "InsertUpdate.Exception.FieldRequired",meta.getKeyStream2()[i])); //$NON-NLS-1$ //$NON-NLS-2$
-                }
-                keynrs2.add(keynr2);
-                
-                if (log.isDebug()) logDebug(BaseMessages.getString(PKG, "InsertUpdate.Log.FieldHasDataNumbers",meta.getKeyStream()[i])+""+keynrs.get(keynrs.size()-1)); //$NON-NLS-1$ //$NON-NLS-2$
-            }
-            
-            data.keynrs  = ArrayUtils.toPrimitive(keynrs.toArray(new Integer[0]));
-            data.keynrs2 = ArrayUtils.toPrimitive(keynrs2.toArray(new Integer[0]));            
-            
-            // Cache the position of the compare fields in Row row
-            //
-            data.valuenrs = new int[meta.getUpdateLookup().length];
-            for (int i=0;i<meta.getUpdateLookup().length;i++)
-            {
-                data.valuenrs[i]=getInputRowMeta().indexOfValue(meta.getUpdateStream()[i]);
-                if (data.valuenrs[i]<0)  // couldn't find field!
-                {
-                    throw new KettleStepException(BaseMessages.getString(PKG, "InsertUpdate.Exception.FieldRequired",meta.getUpdateStream()[i])); //$NON-NLS-1$ //$NON-NLS-2$
-                }
-                if (log.isDebug()) logDebug(BaseMessages.getString(PKG, "InsertUpdate.Log.FieldHasDataNumbers",meta.getUpdateStream()[i])+data.valuenrs[i]); //$NON-NLS-1$ //$NON-NLS-2$
-            }
-            
-            setLookup(getInputRowMeta());
-            
-            data.insertRowMeta = new RowMeta();
-            
-            // Insert the update fields: just names.  Type doesn't matter!
-            for (int i=0;i<meta.getUpdateLookup().length;i++) 
-            {
-                ValueMetaInterface insValue = data.insertRowMeta.searchValueMeta( meta.getUpdateLookup()[i]); 
-                if (insValue==null) // Don't add twice!
-                {
-                    // we already checked that this value exists so it's probably safe to ignore lookup failure...
-                    ValueMetaInterface insertValue = getInputRowMeta().searchValueMeta( meta.getUpdateStream()[i] ).clone();
-                    insertValue.setName(meta.getUpdateLookup()[i]);
-                    data.insertRowMeta.addValueMeta( insertValue );
-                }
-                else
-                {
-                    throw new KettleStepException("The same column can't be inserted into the target row twice: "+insValue.getName()); // TODO i18n
-                }
-            }
-            data.db.prepareInsert(data.insertRowMeta, environmentSubstitute(meta.getSchemaName()), 
-            		                                  environmentSubstitute(meta.getTableName()));
-            
-            if (!meta.isUpdateBypassed())
-            {
-                List<String> updateColumns = new ArrayList<String>();
-                for(int i=0;i<meta.getUpdate().length;i++) {
-                    if(meta.getUpdate()[i].booleanValue()) {
-                        updateColumns.add(meta.getUpdateLookup()[i]);
-                    }
-                }
-                prepareUpdate(getInputRowMeta());
-            }
+            first=false;
+            if (log.isDebug()) logDebug(BaseMessages.getString(PKG, "InsertUpdate.Log.CheckingRow")+getInputRowMeta().getString(r)); //$NON-NLS-1$
+            meta.initializeData(this, data);
         }
 
-		    
 		try
 		{
 			lookupValues(getInputRowMeta(), r); // add new values to the row in rowset[0].
@@ -316,162 +217,8 @@
 		}
 	
 		return true;
-	}
-    
-    public void setLookup(RowMetaInterface rowMeta) throws KettleDatabaseException
-    {
-        data.lookupParameterRowMeta = new RowMeta();
-        data.lookupReturnRowMeta = new RowMeta();
-        
-        DatabaseMeta databaseMeta = meta.getDatabaseMeta();
-        
-        String sql = "SELECT ";
+	}
 
-        for (int i = 0; i < meta.getUpdateLookup().length; i++)
-        {
-            if (i != 0) sql += ", ";
-            sql += databaseMeta.quoteField(meta.getUpdateLookup()[i]);
-            data.lookupReturnRowMeta.addValueMeta( rowMeta.searchValueMeta(meta.getUpdateStream()[i]).clone() );
-        }
-
-        sql += " FROM " + data.schemaTable + " WHERE ";
-
-        for (int i = 0; i < meta.getKeyLookup().length; i++)
-        {
-            if (i != 0){
-            	sql += " AND ";
-            } 
-            
-            sql += " ( ( ";
-            
-            sql += databaseMeta.quoteField(meta.getKeyLookup()[i]);
-            if ("BETWEEN".equalsIgnoreCase(meta.getKeyCondition()[i]))
-            {
-                sql += " BETWEEN ? AND ? ";
-                data.lookupParameterRowMeta.addValueMeta( rowMeta.searchValueMeta(meta.getKeyStream()[i]) );
-                data.lookupParameterRowMeta.addValueMeta( rowMeta.searchValueMeta(meta.getKeyStream2()[i]) );
-            }
-            else
-            {
-                if ("IS NULL".equalsIgnoreCase(meta.getKeyCondition()[i]) || "IS NOT NULL".equalsIgnoreCase(meta.getKeyCondition()[i]))
-                {
-                    sql += " " + meta.getKeyCondition()[i] + " ";
-                }
-                else if ("= ~NULL".equalsIgnoreCase(meta.getKeyCondition()[i])){
-                	
-                	sql += " IS NULL AND ";
-                    
-                	if (databaseMeta.requiresCastToVariousForIsNull()) {
-                        sql += " CAST(? AS VARCHAR(256)) IS NULL ";
-                    }
-                    else{
-                        sql += " ? IS NULL ";
-                    }
-                	// null check
-                    data.lookupParameterRowMeta.addValueMeta( rowMeta.searchValueMeta(meta.getKeyStream()[i]) );
-                    sql += " ) OR ( "+ databaseMeta.quoteField(meta.getKeyLookup()[i])+" = ? ";
-                    // equality check, cloning so auto-rename because of adding same fieldname does not cause problems
-                    data.lookupParameterRowMeta.addValueMeta( rowMeta.searchValueMeta(meta.getKeyStream()[i]).clone() );
-                	
-                }
-                else
-                {
-                    sql += " " + meta.getKeyCondition()[i] + " ? ";
-                    data.lookupParameterRowMeta.addValueMeta( rowMeta.searchValueMeta(meta.getKeyStream()[i]) );
-                }
-            }
-            sql += " ) ) ";
-        }
-        
-        try
-        {
-            if(log.isDetailed()) logDetailed("Setting preparedStatement to [" + sql + "]");
-            data.prepStatementLookup = data.db.getConnection().prepareStatement(databaseMeta.stripCR(sql));
-        }
-        catch (SQLException ex)
-        {
-            throw new KettleDatabaseException("Unable to prepare statement for SQL statement [" + sql + "]", ex);
-        }
-    }
-    
-    // Lookup certain fields in a table
-    public void prepareUpdate(RowMetaInterface rowMeta) throws KettleDatabaseException
-    {
-        DatabaseMeta databaseMeta = meta.getDatabaseMeta();
-        data.updateParameterRowMeta = new RowMeta();
-        
-        String sql = "UPDATE " + data.schemaTable + Const.CR;
-        sql += "SET ";
-        
-        boolean comma=false;
-        
-        for (int i=0;i<meta.getUpdateLookup().length;i++)
-        {
-    		if ( meta.getUpdate()[i].booleanValue() ) {
-                if (comma) sql+= ",   ";
-                else comma=true;
-                
-                sql += databaseMeta.quoteField(meta.getUpdateLookup()[i]);
-                sql += " = ?" + Const.CR;
-                data.updateParameterRowMeta.addValueMeta( rowMeta.searchValueMeta(meta.getUpdateStream()[i]).clone() );
-    		}
-        }
-
-        sql += "WHERE ";
-
-        for (int i=0;i<meta.getKeyLookup().length;i++)
-        {
-            if (i!=0) sql += "AND   ";
-            sql += " ( ( ";
-            sql += databaseMeta.quoteField(meta.getKeyLookup()[i]);
-            if ("BETWEEN".equalsIgnoreCase(meta.getKeyCondition()[i]))
-            {
-                sql += " BETWEEN ? AND ? ";
-                data.updateParameterRowMeta.addValueMeta( rowMeta.searchValueMeta(meta.getKeyStream()[i]) );
-                data.updateParameterRowMeta.addValueMeta( rowMeta.searchValueMeta(meta.getKeyStream2()[i]) );
-            }
-            else
-            if ("IS NULL".equalsIgnoreCase(meta.getKeyCondition()[i]) || "IS NOT NULL".equalsIgnoreCase(meta.getKeyCondition()[i]))
-            {
-                sql += " "+meta.getKeyCondition()[i]+" ";
-            }
-            else if ("= ~NULL".equalsIgnoreCase(meta.getKeyCondition()[i])){
-            	
-            	sql += " IS NULL AND ";
-                
-            	if (databaseMeta.requiresCastToVariousForIsNull()) {
-                    sql += "CAST(? AS VARCHAR(256)) IS NULL";
-                }
-                else{
-                    sql += "? IS NULL";
-                }
-            	// null check
-                data.updateParameterRowMeta.addValueMeta( rowMeta.searchValueMeta(meta.getKeyStream()[i]) );
-                sql += " ) OR ( "+ databaseMeta.quoteField(meta.getKeyLookup()[i])+" = ?";
-                // equality check, cloning so auto-rename because of adding same fieldname does not cause problems
-                data.updateParameterRowMeta.addValueMeta( rowMeta.searchValueMeta(meta.getKeyStream()[i]).clone() );
-            	
-            }              
-            else
-            {
-                sql += " "+meta.getKeyCondition()[i]+" ? ";
-                data.updateParameterRowMeta.addValueMeta( rowMeta.searchValueMeta(meta.getKeyStream()[i]).clone() );
-            }
-            sql += " ) ) ";
-        }
-
-        try
-        {
-        	if(log.isDetailed()) logDetailed("Setting update preparedStatement to ["+sql+"]");
-            data.prepStatementUpdate=data.db.getConnection().prepareStatement(databaseMeta.stripCR(sql));
-        }
-        catch(SQLException ex) 
-        {
-            throw new KettleDatabaseException("Unable to prepare statement for SQL statement [" + sql + "]", ex);
-        }
-    }
-
-	
     public boolean init(StepMetaInterface smi, StepDataInterface sdi)
 	{
 		meta=(InsertUpdateMeta)smi;
diff -r 35f3652aa7b3 -r 3402ecb6225c src/org/pentaho/di/trans/steps/insertupdate/InsertUpdateData.java
--- a/src/org/pentaho/di/trans/steps/insertupdate/InsertUpdateData.java	Wed Aug 10 21:38:44 2011 +0200
+++ b/src/org/pentaho/di/trans/steps/insertupdate/InsertUpdateData.java	Wed Aug 10 21:38:47 2011 +0200
@@ -11,12 +11,12 @@
  
 package org.pentaho.di.trans.steps.insertupdate;
 
-import java.sql.PreparedStatement;
-
-import org.pentaho.di.core.database.Database;
-import org.pentaho.di.core.row.RowMetaInterface;
-import org.pentaho.di.trans.step.BaseStepData;
-import org.pentaho.di.trans.step.StepDataInterface;
+import java.sql.PreparedStatement;
+
+import org.pentaho.di.core.database.Database;
+import org.pentaho.di.core.row.RowMetaInterface;
+import org.pentaho.di.trans.step.BaseStepData;
+import org.pentaho.di.trans.step.StepDataInterface;
 
 
 
@@ -35,7 +35,9 @@
 	public int    valuenrs[];       // Stream valuename nrs to prevent searches.
 
     public RowMetaInterface outputRowMeta;
-
+
+    public String schemaName;
+    public String tableName;
     public String schemaTable;
 
     public PreparedStatement prepStatementLookup;
diff -r 35f3652aa7b3 -r 3402ecb6225c src/org/pentaho/di/trans/steps/insertupdate/InsertUpdateMeta.java
--- a/src/org/pentaho/di/trans/steps/insertupdate/InsertUpdateMeta.java	Wed Aug 10 21:38:44 2011 +0200
+++ b/src/org/pentaho/di/trans/steps/insertupdate/InsertUpdateMeta.java	Wed Aug 10 21:38:47 2011 +0200
@@ -11,9 +11,12 @@
 
 package org.pentaho.di.trans.steps.insertupdate;
 
+import java.sql.SQLException;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 
+import org.apache.commons.lang.ArrayUtils;
 import org.pentaho.di.core.CheckResult;
 import org.pentaho.di.core.CheckResultInterface;
 import org.pentaho.di.core.Const;
@@ -21,8 +24,10 @@
 import org.pentaho.di.core.SQLStatement;
 import org.pentaho.di.core.database.Database;
 import org.pentaho.di.core.database.DatabaseMeta;
+import org.pentaho.di.core.exception.KettleDatabaseException;
 import org.pentaho.di.core.exception.KettleException;
 import org.pentaho.di.core.exception.KettleStepException;
+import org.pentaho.di.core.exception.KettleValueException;
 import org.pentaho.di.core.exception.KettleXMLException;
 import org.pentaho.di.core.row.RowMeta;
 import org.pentaho.di.core.row.RowMetaInterface;
@@ -30,8 +35,8 @@
 import org.pentaho.di.core.variables.VariableSpace;
 import org.pentaho.di.core.xml.XMLHandler;
 import org.pentaho.di.i18n.BaseMessages;
-import org.pentaho.di.repository.Repository;
 import org.pentaho.di.repository.ObjectId;
+import org.pentaho.di.repository.Repository;
 import org.pentaho.di.shared.SharedObjectInterface;
 import org.pentaho.di.trans.DatabaseImpact;
 import org.pentaho.di.trans.Trans;
@@ -485,10 +490,6 @@
 		}
 	}
 	
-	public void getFields(RowMetaInterface rowMeta, String origin, RowMetaInterface[] info, StepMeta nextStep, VariableSpace space) throws KettleStepException
-	{
-		// Default: nothing changes to rowMeta
-	}
 
 	public void check(List<CheckResultInterface> remarks, TransMeta transMeta, StepMeta stepMeta, RowMetaInterface prev, String input[], String output[], RowMetaInterface info)
 	{
@@ -948,5 +949,277 @@
     public boolean supportsErrorHandling()
     {
         return true;
-    }
-}
\ No newline at end of file
+    }
+
+
+    /** Perform lazy initialization from meta data
+     * 
+     * 
+     * @param step TODO
+     * @param data TODO
+     * 
+     * @throws KettleStepException
+     * @throws KettleValueException
+     * @throws KettleDatabaseException
+     */
+    void initializeData(InsertUpdate step, InsertUpdateData data) 
+            throws KettleStepException, KettleValueException, KettleDatabaseException
+    {
+        RowMetaInterface inputRowMeta = step.getInputRowMeta();
+
+        data.schemaName = step.environmentSubstitute(schemaName);
+        data.tableName = step.environmentSubstitute(tableName);
+        data.schemaTable = databaseMeta.getQuotedSchemaTableCombination(data.schemaName, data.tableName);
+
+        data.outputRowMeta = inputRowMeta.clone();
+        getFields(data.outputRowMeta, step.getStepname(), null, null, step);
+
+        readKeyNrs(data, inputRowMeta);
+        cacheCompareFieldsRowPosition(data, inputRowMeta);
+        setLookup(data, inputRowMeta);
+        prepareInsert(data, inputRowMeta);
+        prepareUpdateStatement(data, inputRowMeta);
+    }
+
+
+    private void readKeyNrs(InsertUpdateData data, RowMetaInterface inputRowMeta) throws KettleStepException
+    {
+        ArrayList<Integer> keynrs = new ArrayList<Integer>(keyStream.length);
+        ArrayList<Integer> keynrs2 = new ArrayList<Integer>(keyStream.length);
+
+        for (int i = 0; i < keyStream.length; i++) {
+            int keynr = inputRowMeta.indexOfValue(keyStream[i]);
+
+            if (keynr < 0 && // couldn't find field!
+                    !"IS NULL".equalsIgnoreCase(keyCondition[i]) && // No field needed! //$NON-NLS-1$
+                    !"IS NOT NULL".equalsIgnoreCase(keyCondition[i]) // No field needed! //$NON-NLS-1$
+            ) {
+                throw new KettleStepException(BaseMessages.getString(PKG, "InsertUpdate.Exception.FieldRequired", keyStream[i])); //$NON-NLS-1$ //$NON-NLS-2$
+            }
+            keynrs.add(keynr);
+
+            // this operator needs two bindings
+            if ("= ~NULL".equalsIgnoreCase(keyCondition[i])) {
+                keynrs.add(keynr);
+                keynrs2.add(-1);
+            }
+
+            int keynr2 = inputRowMeta.indexOfValue(keyStream2[i]);
+            if (keynr2 < 0 && // couldn't find field!
+                    "BETWEEN".equalsIgnoreCase(keyCondition[i]) // 2 fields needed! //$NON-NLS-1$
+            ) {
+                throw new KettleStepException(BaseMessages.getString(PKG, "InsertUpdate.Exception.FieldRequired", keyStream2[i])); //$NON-NLS-1$ //$NON-NLS-2$
+            }
+            keynrs2.add(keynr2);
+
+            // TODO: the logging moved from BaseStep.log to BaseStepMeta.log
+            // if (log.isDebug())
+            //                logDebug(BaseMessages.getString(PKG, "InsertUpdate.Log.FieldHasDataNumbers", keyStream[i]) + "" + keynrs.get(keynrs.size() - 1)); //$NON-NLS-1$ //$NON-NLS-2$
+        }
+
+        data.keynrs = ArrayUtils.toPrimitive(keynrs.toArray(new Integer[0]));
+        data.keynrs2 = ArrayUtils.toPrimitive(keynrs2.toArray(new Integer[0]));
+    }
+
+    private void cacheCompareFieldsRowPosition(InsertUpdateData data, RowMetaInterface inputRowMeta) throws KettleStepException {
+        data.valuenrs = new int[updateLookup.length];
+        for (int i = 0; i < updateLookup.length; i++)
+        {
+            data.valuenrs[i] = inputRowMeta.indexOfValue(updateStream[i]);
+            if (data.valuenrs[i]<0)  // couldn't find field!
+            {
+                throw new KettleStepException(BaseMessages.getString(PKG,
+                        "InsertUpdate.Exception.FieldRequired", updateStream[i])); //$NON-NLS-1$ //$NON-NLS-2$
+            }
+            // TODO: fix logging
+            //            if (log.isDebug()) insertUpdate.logDebug(BaseMessages.getString(InsertUpdate.PKG, "InsertUpdate.Log.FieldHasDataNumbers",getUpdateStream()[i])+data.valuenrs[i]); //$NON-NLS-1$ //$NON-NLS-2$
+        }
+    }
+
+    private void setLookup(InsertUpdateData data, RowMetaInterface rowMeta) throws KettleDatabaseException
+    {
+        data.lookupParameterRowMeta = new RowMeta();
+        data.lookupReturnRowMeta = new RowMeta();
+
+        String sql = "SELECT ";
+
+        for (int i = 0; i < updateLookup.length; i++)
+        {
+            if (i != 0) sql += ", ";
+            sql += databaseMeta.quoteField(updateLookup[i]);
+            data.lookupReturnRowMeta.addValueMeta( rowMeta.searchValueMeta(updateStream[i]).clone() );
+        }
+
+        sql += " FROM " + data.schemaTable + " WHERE ";
+
+        for (int i = 0; i < keyLookup.length; i++)
+        {
+            if (i != 0) {
+                sql += " AND ";
+            }
+
+            sql += " ( ( ";
+            sql += databaseMeta.quoteField(keyLookup[i]);
+            if ("BETWEEN".equalsIgnoreCase(keyCondition[i]))
+            {
+                sql += " BETWEEN ? AND ? ";
+                data.lookupParameterRowMeta.addValueMeta( rowMeta.searchValueMeta(keyStream[i]) );
+                data.lookupParameterRowMeta.addValueMeta( rowMeta.searchValueMeta(keyStream2[i]) );
+            }
+            else
+            {
+                if ("IS NULL".equalsIgnoreCase(keyCondition[i]) || "IS NOT NULL".equalsIgnoreCase(keyCondition[i]))
+                {
+                    sql += " " + keyCondition[i] + " ";
+                }
+                else if ("= ~NULL".equalsIgnoreCase(keyCondition[i])){
+
+                    sql += " IS NULL AND ";
+
+                    if (databaseMeta.requiresCastToVariousForIsNull()) {
+                        sql += " CAST(? AS VARCHAR(256)) IS NULL ";
+                    }
+                    else{
+                        sql += " ? IS NULL ";
+                    }
+                    // null check
+                    data.lookupParameterRowMeta.addValueMeta( rowMeta.searchValueMeta(keyStream[i]) );
+                    sql += " ) OR ( "+ databaseMeta.quoteField(keyLookup[i])+" = ? ";
+                    // equality check, cloning so auto-rename because of adding same fieldname does not cause problems
+                    data.lookupParameterRowMeta.addValueMeta( rowMeta.searchValueMeta(keyStream[i]).clone() );
+
+                }
+                else
+                {
+                    sql += " " + keyCondition[i] + " ? ";
+                    data.lookupParameterRowMeta.addValueMeta( rowMeta.searchValueMeta(keyStream[i]) );
+                }
+            }
+            sql += " ) ) ";
+        }
+
+        try
+        {
+            // TODO fix logging
+            // if(log.isDetailed()) insertUpdate.logDetailed("Setting preparedStatement to [" + sql + "]");
+            data.prepStatementLookup = data.db.getConnection().prepareStatement(databaseMeta.stripCR(sql));
+        }
+        catch (SQLException ex)
+        {
+            throw new KettleDatabaseException("Unable to prepare statement for SQL statement [" + sql + "]", ex);
+        }
+    }
+
+
+    private void prepareInsert(InsertUpdateData data, RowMetaInterface inputRowMeta) throws KettleStepException,
+            KettleDatabaseException {
+        data.insertRowMeta = new RowMeta();
+
+        // Insert the update fields: just names.  Type doesn't matter!
+        for (int i = 0; i < updateLookup.length; i++)
+        {
+            ValueMetaInterface insValue = data.insertRowMeta.searchValueMeta(updateLookup[i]);
+            if (insValue==null) // Don't add twice!
+            {
+                // we already checked that this value exists so it's probably safe to ignore lookup failure...
+                ValueMetaInterface insertValue = inputRowMeta.searchValueMeta(updateStream[i]).clone();
+                insertValue.setName(updateLookup[i]);
+                data.insertRowMeta.addValueMeta( insertValue );
+            }
+            else
+            {
+                throw new KettleStepException("The same column can't be inserted into the target row twice: "+insValue.getName()); // TODO i18n
+            }
+        }
+
+        data.db.prepareInsert(data.insertRowMeta, data.schemaName, data.tableName);
+    }
+
+
+    private void prepareUpdateStatement(InsertUpdateData data, RowMetaInterface inputRowMeta) throws KettleDatabaseException
+    {
+        if (!isUpdateBypassed())
+        {
+            List<String> updateColumns = new ArrayList<String>();
+            for (int i = 0; i < update.length; i++) {
+                if (update[i].booleanValue()) {
+                    updateColumns.add(updateLookup[i]);
+                }
+            }
+            prepareUpdate(data, inputRowMeta);
+        }
+    }
+
+
+    private void prepareUpdate(InsertUpdateData data, RowMetaInterface rowMeta) throws KettleDatabaseException
+    {
+        data.updateParameterRowMeta = new RowMeta();
+
+        String sql = "UPDATE " + data.schemaTable + Const.CR;
+        sql += "SET ";
+
+        boolean comma=false;
+
+        for (int i = 0; i < updateLookup.length; i++)
+        {
+            if (update[i].booleanValue()) {
+                if (comma) sql+= ",   ";
+                else comma=true;
+
+                sql += databaseMeta.quoteField(updateLookup[i]);
+                sql += " = ?" + Const.CR;
+                data.updateParameterRowMeta.addValueMeta(rowMeta.searchValueMeta(updateStream[i]).clone());
+            }
+        }
+
+        sql += "WHERE ";
+
+        for (int i = 0; i < keyLookup.length; i++)
+        {
+            if (i!=0) sql += "AND   ";
+            sql += " ( ( ";
+            sql += databaseMeta.quoteField(keyLookup[i]);
+            if ("BETWEEN".equalsIgnoreCase(keyCondition[i]))
+            {
+                sql += " BETWEEN ? AND ? ";
+                data.updateParameterRowMeta.addValueMeta(rowMeta.searchValueMeta(keyStream[i]));
+                data.updateParameterRowMeta.addValueMeta(rowMeta.searchValueMeta(keyStream2[i]));
+            }
+            else
+                if ("IS NULL".equalsIgnoreCase(keyCondition[i]) || "IS NOT NULL".equalsIgnoreCase(keyCondition[i]))
+                {
+                    sql += " " + keyCondition[i] + " ";
+                }
+                else
+                    if ("= ~NULL".equalsIgnoreCase(keyCondition[i])) {
+                        sql += " IS NULL AND ";
+                        if (databaseMeta.requiresCastToVariousForIsNull()) {
+                            sql += "CAST(? AS VARCHAR(256)) IS NULL";
+                        } else {
+                            sql += "? IS NULL";
+                        }
+                        // null check
+                        data.updateParameterRowMeta.addValueMeta(rowMeta.searchValueMeta(keyStream[i]));
+                        sql += " ) OR ( " + databaseMeta.quoteField(keyLookup[i]) + " = ?";
+                        // equality check, cloning so auto-rename because of adding same fieldname does not cause problems
+                        data.updateParameterRowMeta.addValueMeta(rowMeta.searchValueMeta(keyStream[i]).clone());
+
+                    } else {
+                        sql += " " + keyCondition[i] + " ? ";
+                        data.updateParameterRowMeta.addValueMeta(rowMeta.searchValueMeta(keyStream[i]).clone());
+                    }
+            sql += " ) ) ";
+        }
+
+        try
+        {
+            // TODO fix loggin
+            // if(log.isDetailed()) insertUpdate.logDetailed("Setting update preparedStatement to ["+sql+"]");
+            data.prepStatementUpdate=data.db.getConnection().prepareStatement(databaseMeta.stripCR(sql));
+        }
+        catch (SQLException ex)
+        {
+            throw new KettleDatabaseException("Unable to prepare statement for SQL statement [" + sql + "]", ex);
+        }
+    }
+}
