@Test public void testWriteToTable() throws Exception { tableOutputSpy.writeToTable( mock( RowMetaInterface.class ), new Object[]{} ); }
protected Object[] writeToTable( RowMetaInterface rowMeta, Object[] r ) throws KettleException { if ( r == null ) { 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() ) { 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"; logError( 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() ) { 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 ) ) { 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() ) { insertRowData = new Object[data.valuenrs.length]; for ( int idx = 0; idx < data.valuenrs.length; idx++ ) { insertRowData[idx] = r[data.valuenrs[idx]]; } } if ( Utils.isEmpty( tableName ) ) { throw new KettleStepException( "The tablename is not defined (empty)" ); } insertStatement = data.preparedStatements.get( tableName ); if ( insertStatement == null ) { String sql = data.db .getInsertStatement( environmentSubstitute( meta.getSchemaName() ), tableName, data.insertRowMeta ); if ( log.isDetailed() ) { logDetailed( "Prepared statement : " + sql ); } if(meta.getDatabaseMeta().getURL().contains("phoenix")){ sql=sql.replace("insert", "upsert"); sql=sql.replace("INSERT", "UPSERT"); sql=sql.replace("update", "upsert"); sql=sql.replace("UPDATE", "UPSERT"); } try { insertStatement = data.db.prepareSQL( sql, meta.isReturningGeneratedKeys() ); } catch ( Exception e ) { logError( e.toString() ); } data.preparedStatements.put( tableName, insertStatement ); } try { if ( data.useSafePoints ) { data.savepoint = data.db.setSavepoint(); } data.db.setValues( data.insertRowMeta, insertRowData, insertStatement ); data.db.insertRow( insertStatement, data.batchMode, false ); if ( isRowLevel() ) { logRowlevel( "Written row: " + data.insertRowMeta.getString( insertRowData ) ); } Integer commitCounter = data.commitCounterMap.get( tableName ); if ( commitCounter == null ) { commitCounter = Integer.valueOf( 1 ); } else { commitCounter++; } data.commitCounterMap.put( tableName, Integer.valueOf( commitCounter.intValue() ) ); if ( data.useSafePoints ) { if ( data.releaseSavepoint ) { data.db.releaseSavepoint( data.savepoint ); } } if ( ( data.commitSize > 0 ) && ( ( commitCounter % data.commitSize ) == 0 ) ) { if ( data.db.getUseBatchInsert( data.batchMode ) ) { try { insertStatement.executeBatch(); data.db.commit(); insertStatement.clearBatch(); } catch ( SQLException ex ) { throw Database.createKettleDatabaseBatchException( "Error updating batch", ex ); } catch ( Exception ex ) { throw new KettleDatabaseException( "Unexpected error inserting row", ex ); } } else { data.db.commit(); } data.commitCounterMap.put( tableName, Integer.valueOf( 0 ) ); rowIsSafe = true; } else { rowIsSafe = false; } if ( meta.isReturningGeneratedKeys() ) { RowMetaAndData extraKeys = data.db.getGeneratedKeys( insertStatement ); if ( extraKeys.getRowMeta().size() > 0 ) { generatedKey = extraKeys.getRowMeta().getInteger( extraKeys.getData(), 0 ); } else { throw new KettleStepException( "No generated keys while \"return generated keys\" is active!" ); } } } catch ( KettleDatabaseBatchException be ) { errorMessage = be.toString(); batchProblem = true; sendToErrorRow = true; updateCounts = be.getUpdateCounts(); exceptionsList = be.getExceptionsList(); if ( getStepMeta().isDoingErrorHandling() ) { data.db.clearBatch( insertStatement ); data.db.commit( true ); } else { data.db.clearBatch( insertStatement ); data.db.rollback(); StringBuilder msg = new StringBuilder( "Error batch inserting rows into table [" + tableName + "]." ); msg.append( Const.CR ); msg.append( "Errors encountered (first 10):" ).append( Const.CR ); for ( int x = 0; x < be.getExceptionsList().size() && x < 10; x++ ) { Exception exception = be.getExceptionsList().get( x ); if ( exception.getMessage() != null ) { msg.append( exception.getMessage() ).append( Const.CR ); } } throw new KettleException( msg.toString(), be ); } } catch ( KettleDatabaseException dbe ) { if ( getStepMeta().isDoingErrorHandling() ) { if ( isRowLevel() ) { logRowlevel( "Written row to error handling : " + getInputRowMeta().getString( r ) ); } if ( data.useSafePoints ) { data.db.rollback( data.savepoint ); if ( data.releaseSavepoint ) { data.db.releaseSavepoint( data.savepoint ); } } sendToErrorRow = true; errorMessage = dbe.toString(); } else { if ( meta.ignoreErrors() ) { if ( data.warnings < 20 ) { if ( log.isBasic() ) { logBasic( "WARNING: Couldn't insert row into table: " + rowMeta.getString( r ) + Const.CR + dbe.getMessage() ); } } else if ( data.warnings == 20 ) { if ( log.isBasic() ) { logBasic( "FINAL WARNING (no more then 20 displayed): Couldn't insert row into table: " + rowMeta.getString( r ) + Const.CR + dbe.getMessage() ); } } data.warnings++; } else { setErrors( getErrors() + 1 ); data.db.rollback(); throw new KettleException( "Error inserting row into table [" + tableName + "] with values: " + rowMeta.getString( r ), dbe ); } } } if ( generatedKey != null ) { outputRowData = RowDataUtil.addValueData( outputRowData, rowMeta.size(), generatedKey ); } if ( data.batchMode ) { if ( sendToErrorRow ) { if ( batchProblem ) { data.batchBuffer.add( outputRowData ); outputRowData = null; processBatchException( errorMessage, updateCounts, exceptionsList ); } else { putError( rowMeta, r, 1L, errorMessage, null, "TOP001" ); outputRowData = null; } } else { data.batchBuffer.add( outputRowData ); outputRowData = null; if ( rowIsSafe ) { for ( int i = 0; i < data.batchBuffer.size(); i++ ) { Object[] row = data.batchBuffer.get( i ); putRow( data.outputRowMeta, row ); incrementLinesOutput(); } data.batchBuffer.clear(); } } } else { if ( sendToErrorRow ) { putError( rowMeta, r, 1, errorMessage, null, "TOP001" ); outputRowData = null; } } return outputRowData; }
TableOutput extends BaseStep implements StepInterface { protected Object[] writeToTable( RowMetaInterface rowMeta, Object[] r ) throws KettleException { if ( r == null ) { 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() ) { 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"; logError( 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() ) { 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 ) ) { 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() ) { insertRowData = new Object[data.valuenrs.length]; for ( int idx = 0; idx < data.valuenrs.length; idx++ ) { insertRowData[idx] = r[data.valuenrs[idx]]; } } if ( Utils.isEmpty( tableName ) ) { throw new KettleStepException( "The tablename is not defined (empty)" ); } insertStatement = data.preparedStatements.get( tableName ); if ( insertStatement == null ) { String sql = data.db .getInsertStatement( environmentSubstitute( meta.getSchemaName() ), tableName, data.insertRowMeta ); if ( log.isDetailed() ) { logDetailed( "Prepared statement : " + sql ); } if(meta.getDatabaseMeta().getURL().contains("phoenix")){ sql=sql.replace("insert", "upsert"); sql=sql.replace("INSERT", "UPSERT"); sql=sql.replace("update", "upsert"); sql=sql.replace("UPDATE", "UPSERT"); } try { insertStatement = data.db.prepareSQL( sql, meta.isReturningGeneratedKeys() ); } catch ( Exception e ) { logError( e.toString() ); } data.preparedStatements.put( tableName, insertStatement ); } try { if ( data.useSafePoints ) { data.savepoint = data.db.setSavepoint(); } data.db.setValues( data.insertRowMeta, insertRowData, insertStatement ); data.db.insertRow( insertStatement, data.batchMode, false ); if ( isRowLevel() ) { logRowlevel( "Written row: " + data.insertRowMeta.getString( insertRowData ) ); } Integer commitCounter = data.commitCounterMap.get( tableName ); if ( commitCounter == null ) { commitCounter = Integer.valueOf( 1 ); } else { commitCounter++; } data.commitCounterMap.put( tableName, Integer.valueOf( commitCounter.intValue() ) ); if ( data.useSafePoints ) { if ( data.releaseSavepoint ) { data.db.releaseSavepoint( data.savepoint ); } } if ( ( data.commitSize > 0 ) && ( ( commitCounter % data.commitSize ) == 0 ) ) { if ( data.db.getUseBatchInsert( data.batchMode ) ) { try { insertStatement.executeBatch(); data.db.commit(); insertStatement.clearBatch(); } catch ( SQLException ex ) { throw Database.createKettleDatabaseBatchException( "Error updating batch", ex ); } catch ( Exception ex ) { throw new KettleDatabaseException( "Unexpected error inserting row", ex ); } } else { data.db.commit(); } data.commitCounterMap.put( tableName, Integer.valueOf( 0 ) ); rowIsSafe = true; } else { rowIsSafe = false; } if ( meta.isReturningGeneratedKeys() ) { RowMetaAndData extraKeys = data.db.getGeneratedKeys( insertStatement ); if ( extraKeys.getRowMeta().size() > 0 ) { generatedKey = extraKeys.getRowMeta().getInteger( extraKeys.getData(), 0 ); } else { throw new KettleStepException( "No generated keys while \"return generated keys\" is active!" ); } } } catch ( KettleDatabaseBatchException be ) { errorMessage = be.toString(); batchProblem = true; sendToErrorRow = true; updateCounts = be.getUpdateCounts(); exceptionsList = be.getExceptionsList(); if ( getStepMeta().isDoingErrorHandling() ) { data.db.clearBatch( insertStatement ); data.db.commit( true ); } else { data.db.clearBatch( insertStatement ); data.db.rollback(); StringBuilder msg = new StringBuilder( "Error batch inserting rows into table [" + tableName + "]." ); msg.append( Const.CR ); msg.append( "Errors encountered (first 10):" ).append( Const.CR ); for ( int x = 0; x < be.getExceptionsList().size() && x < 10; x++ ) { Exception exception = be.getExceptionsList().get( x ); if ( exception.getMessage() != null ) { msg.append( exception.getMessage() ).append( Const.CR ); } } throw new KettleException( msg.toString(), be ); } } catch ( KettleDatabaseException dbe ) { if ( getStepMeta().isDoingErrorHandling() ) { if ( isRowLevel() ) { logRowlevel( "Written row to error handling : " + getInputRowMeta().getString( r ) ); } if ( data.useSafePoints ) { data.db.rollback( data.savepoint ); if ( data.releaseSavepoint ) { data.db.releaseSavepoint( data.savepoint ); } } sendToErrorRow = true; errorMessage = dbe.toString(); } else { if ( meta.ignoreErrors() ) { if ( data.warnings < 20 ) { if ( log.isBasic() ) { logBasic( "WARNING: Couldn't insert row into table: " + rowMeta.getString( r ) + Const.CR + dbe.getMessage() ); } } else if ( data.warnings == 20 ) { if ( log.isBasic() ) { logBasic( "FINAL WARNING (no more then 20 displayed): Couldn't insert row into table: " + rowMeta.getString( r ) + Const.CR + dbe.getMessage() ); } } data.warnings++; } else { setErrors( getErrors() + 1 ); data.db.rollback(); throw new KettleException( "Error inserting row into table [" + tableName + "] with values: " + rowMeta.getString( r ), dbe ); } } } if ( generatedKey != null ) { outputRowData = RowDataUtil.addValueData( outputRowData, rowMeta.size(), generatedKey ); } if ( data.batchMode ) { if ( sendToErrorRow ) { if ( batchProblem ) { data.batchBuffer.add( outputRowData ); outputRowData = null; processBatchException( errorMessage, updateCounts, exceptionsList ); } else { putError( rowMeta, r, 1L, errorMessage, null, "TOP001" ); outputRowData = null; } } else { data.batchBuffer.add( outputRowData ); outputRowData = null; if ( rowIsSafe ) { for ( int i = 0; i < data.batchBuffer.size(); i++ ) { Object[] row = data.batchBuffer.get( i ); putRow( data.outputRowMeta, row ); incrementLinesOutput(); } data.batchBuffer.clear(); } } } else { if ( sendToErrorRow ) { putError( rowMeta, r, 1, errorMessage, null, "TOP001" ); outputRowData = null; } } return outputRowData; } }
TableOutput extends BaseStep implements StepInterface { protected Object[] writeToTable( RowMetaInterface rowMeta, Object[] r ) throws KettleException { if ( r == null ) { 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() ) { 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"; logError( 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() ) { 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 ) ) { 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() ) { insertRowData = new Object[data.valuenrs.length]; for ( int idx = 0; idx < data.valuenrs.length; idx++ ) { insertRowData[idx] = r[data.valuenrs[idx]]; } } if ( Utils.isEmpty( tableName ) ) { throw new KettleStepException( "The tablename is not defined (empty)" ); } insertStatement = data.preparedStatements.get( tableName ); if ( insertStatement == null ) { String sql = data.db .getInsertStatement( environmentSubstitute( meta.getSchemaName() ), tableName, data.insertRowMeta ); if ( log.isDetailed() ) { logDetailed( "Prepared statement : " + sql ); } if(meta.getDatabaseMeta().getURL().contains("phoenix")){ sql=sql.replace("insert", "upsert"); sql=sql.replace("INSERT", "UPSERT"); sql=sql.replace("update", "upsert"); sql=sql.replace("UPDATE", "UPSERT"); } try { insertStatement = data.db.prepareSQL( sql, meta.isReturningGeneratedKeys() ); } catch ( Exception e ) { logError( e.toString() ); } data.preparedStatements.put( tableName, insertStatement ); } try { if ( data.useSafePoints ) { data.savepoint = data.db.setSavepoint(); } data.db.setValues( data.insertRowMeta, insertRowData, insertStatement ); data.db.insertRow( insertStatement, data.batchMode, false ); if ( isRowLevel() ) { logRowlevel( "Written row: " + data.insertRowMeta.getString( insertRowData ) ); } Integer commitCounter = data.commitCounterMap.get( tableName ); if ( commitCounter == null ) { commitCounter = Integer.valueOf( 1 ); } else { commitCounter++; } data.commitCounterMap.put( tableName, Integer.valueOf( commitCounter.intValue() ) ); if ( data.useSafePoints ) { if ( data.releaseSavepoint ) { data.db.releaseSavepoint( data.savepoint ); } } if ( ( data.commitSize > 0 ) && ( ( commitCounter % data.commitSize ) == 0 ) ) { if ( data.db.getUseBatchInsert( data.batchMode ) ) { try { insertStatement.executeBatch(); data.db.commit(); insertStatement.clearBatch(); } catch ( SQLException ex ) { throw Database.createKettleDatabaseBatchException( "Error updating batch", ex ); } catch ( Exception ex ) { throw new KettleDatabaseException( "Unexpected error inserting row", ex ); } } else { data.db.commit(); } data.commitCounterMap.put( tableName, Integer.valueOf( 0 ) ); rowIsSafe = true; } else { rowIsSafe = false; } if ( meta.isReturningGeneratedKeys() ) { RowMetaAndData extraKeys = data.db.getGeneratedKeys( insertStatement ); if ( extraKeys.getRowMeta().size() > 0 ) { generatedKey = extraKeys.getRowMeta().getInteger( extraKeys.getData(), 0 ); } else { throw new KettleStepException( "No generated keys while \"return generated keys\" is active!" ); } } } catch ( KettleDatabaseBatchException be ) { errorMessage = be.toString(); batchProblem = true; sendToErrorRow = true; updateCounts = be.getUpdateCounts(); exceptionsList = be.getExceptionsList(); if ( getStepMeta().isDoingErrorHandling() ) { data.db.clearBatch( insertStatement ); data.db.commit( true ); } else { data.db.clearBatch( insertStatement ); data.db.rollback(); StringBuilder msg = new StringBuilder( "Error batch inserting rows into table [" + tableName + "]." ); msg.append( Const.CR ); msg.append( "Errors encountered (first 10):" ).append( Const.CR ); for ( int x = 0; x < be.getExceptionsList().size() && x < 10; x++ ) { Exception exception = be.getExceptionsList().get( x ); if ( exception.getMessage() != null ) { msg.append( exception.getMessage() ).append( Const.CR ); } } throw new KettleException( msg.toString(), be ); } } catch ( KettleDatabaseException dbe ) { if ( getStepMeta().isDoingErrorHandling() ) { if ( isRowLevel() ) { logRowlevel( "Written row to error handling : " + getInputRowMeta().getString( r ) ); } if ( data.useSafePoints ) { data.db.rollback( data.savepoint ); if ( data.releaseSavepoint ) { data.db.releaseSavepoint( data.savepoint ); } } sendToErrorRow = true; errorMessage = dbe.toString(); } else { if ( meta.ignoreErrors() ) { if ( data.warnings < 20 ) { if ( log.isBasic() ) { logBasic( "WARNING: Couldn't insert row into table: " + rowMeta.getString( r ) + Const.CR + dbe.getMessage() ); } } else if ( data.warnings == 20 ) { if ( log.isBasic() ) { logBasic( "FINAL WARNING (no more then 20 displayed): Couldn't insert row into table: " + rowMeta.getString( r ) + Const.CR + dbe.getMessage() ); } } data.warnings++; } else { setErrors( getErrors() + 1 ); data.db.rollback(); throw new KettleException( "Error inserting row into table [" + tableName + "] with values: " + rowMeta.getString( r ), dbe ); } } } if ( generatedKey != null ) { outputRowData = RowDataUtil.addValueData( outputRowData, rowMeta.size(), generatedKey ); } if ( data.batchMode ) { if ( sendToErrorRow ) { if ( batchProblem ) { data.batchBuffer.add( outputRowData ); outputRowData = null; processBatchException( errorMessage, updateCounts, exceptionsList ); } else { putError( rowMeta, r, 1L, errorMessage, null, "TOP001" ); outputRowData = null; } } else { data.batchBuffer.add( outputRowData ); outputRowData = null; if ( rowIsSafe ) { for ( int i = 0; i < data.batchBuffer.size(); i++ ) { Object[] row = data.batchBuffer.get( i ); putRow( data.outputRowMeta, row ); incrementLinesOutput(); } data.batchBuffer.clear(); } } } else { if ( sendToErrorRow ) { putError( rowMeta, r, 1, errorMessage, null, "TOP001" ); outputRowData = null; } } return outputRowData; } TableOutput( StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta,
    Trans trans ); }
TableOutput extends BaseStep implements StepInterface { protected Object[] writeToTable( RowMetaInterface rowMeta, Object[] r ) throws KettleException { if ( r == null ) { 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() ) { 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"; logError( 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() ) { 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 ) ) { 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() ) { insertRowData = new Object[data.valuenrs.length]; for ( int idx = 0; idx < data.valuenrs.length; idx++ ) { insertRowData[idx] = r[data.valuenrs[idx]]; } } if ( Utils.isEmpty( tableName ) ) { throw new KettleStepException( "The tablename is not defined (empty)" ); } insertStatement = data.preparedStatements.get( tableName ); if ( insertStatement == null ) { String sql = data.db .getInsertStatement( environmentSubstitute( meta.getSchemaName() ), tableName, data.insertRowMeta ); if ( log.isDetailed() ) { logDetailed( "Prepared statement : " + sql ); } if(meta.getDatabaseMeta().getURL().contains("phoenix")){ sql=sql.replace("insert", "upsert"); sql=sql.replace("INSERT", "UPSERT"); sql=sql.replace("update", "upsert"); sql=sql.replace("UPDATE", "UPSERT"); } try { insertStatement = data.db.prepareSQL( sql, meta.isReturningGeneratedKeys() ); } catch ( Exception e ) { logError( e.toString() ); } data.preparedStatements.put( tableName, insertStatement ); } try { if ( data.useSafePoints ) { data.savepoint = data.db.setSavepoint(); } data.db.setValues( data.insertRowMeta, insertRowData, insertStatement ); data.db.insertRow( insertStatement, data.batchMode, false ); if ( isRowLevel() ) { logRowlevel( "Written row: " + data.insertRowMeta.getString( insertRowData ) ); } Integer commitCounter = data.commitCounterMap.get( tableName ); if ( commitCounter == null ) { commitCounter = Integer.valueOf( 1 ); } else { commitCounter++; } data.commitCounterMap.put( tableName, Integer.valueOf( commitCounter.intValue() ) ); if ( data.useSafePoints ) { if ( data.releaseSavepoint ) { data.db.releaseSavepoint( data.savepoint ); } } if ( ( data.commitSize > 0 ) && ( ( commitCounter % data.commitSize ) == 0 ) ) { if ( data.db.getUseBatchInsert( data.batchMode ) ) { try { insertStatement.executeBatch(); data.db.commit(); insertStatement.clearBatch(); } catch ( SQLException ex ) { throw Database.createKettleDatabaseBatchException( "Error updating batch", ex ); } catch ( Exception ex ) { throw new KettleDatabaseException( "Unexpected error inserting row", ex ); } } else { data.db.commit(); } data.commitCounterMap.put( tableName, Integer.valueOf( 0 ) ); rowIsSafe = true; } else { rowIsSafe = false; } if ( meta.isReturningGeneratedKeys() ) { RowMetaAndData extraKeys = data.db.getGeneratedKeys( insertStatement ); if ( extraKeys.getRowMeta().size() > 0 ) { generatedKey = extraKeys.getRowMeta().getInteger( extraKeys.getData(), 0 ); } else { throw new KettleStepException( "No generated keys while \"return generated keys\" is active!" ); } } } catch ( KettleDatabaseBatchException be ) { errorMessage = be.toString(); batchProblem = true; sendToErrorRow = true; updateCounts = be.getUpdateCounts(); exceptionsList = be.getExceptionsList(); if ( getStepMeta().isDoingErrorHandling() ) { data.db.clearBatch( insertStatement ); data.db.commit( true ); } else { data.db.clearBatch( insertStatement ); data.db.rollback(); StringBuilder msg = new StringBuilder( "Error batch inserting rows into table [" + tableName + "]." ); msg.append( Const.CR ); msg.append( "Errors encountered (first 10):" ).append( Const.CR ); for ( int x = 0; x < be.getExceptionsList().size() && x < 10; x++ ) { Exception exception = be.getExceptionsList().get( x ); if ( exception.getMessage() != null ) { msg.append( exception.getMessage() ).append( Const.CR ); } } throw new KettleException( msg.toString(), be ); } } catch ( KettleDatabaseException dbe ) { if ( getStepMeta().isDoingErrorHandling() ) { if ( isRowLevel() ) { logRowlevel( "Written row to error handling : " + getInputRowMeta().getString( r ) ); } if ( data.useSafePoints ) { data.db.rollback( data.savepoint ); if ( data.releaseSavepoint ) { data.db.releaseSavepoint( data.savepoint ); } } sendToErrorRow = true; errorMessage = dbe.toString(); } else { if ( meta.ignoreErrors() ) { if ( data.warnings < 20 ) { if ( log.isBasic() ) { logBasic( "WARNING: Couldn't insert row into table: " + rowMeta.getString( r ) + Const.CR + dbe.getMessage() ); } } else if ( data.warnings == 20 ) { if ( log.isBasic() ) { logBasic( "FINAL WARNING (no more then 20 displayed): Couldn't insert row into table: " + rowMeta.getString( r ) + Const.CR + dbe.getMessage() ); } } data.warnings++; } else { setErrors( getErrors() + 1 ); data.db.rollback(); throw new KettleException( "Error inserting row into table [" + tableName + "] with values: " + rowMeta.getString( r ), dbe ); } } } if ( generatedKey != null ) { outputRowData = RowDataUtil.addValueData( outputRowData, rowMeta.size(), generatedKey ); } if ( data.batchMode ) { if ( sendToErrorRow ) { if ( batchProblem ) { data.batchBuffer.add( outputRowData ); outputRowData = null; processBatchException( errorMessage, updateCounts, exceptionsList ); } else { putError( rowMeta, r, 1L, errorMessage, null, "TOP001" ); outputRowData = null; } } else { data.batchBuffer.add( outputRowData ); outputRowData = null; if ( rowIsSafe ) { for ( int i = 0; i < data.batchBuffer.size(); i++ ) { Object[] row = data.batchBuffer.get( i ); putRow( data.outputRowMeta, row ); incrementLinesOutput(); } data.batchBuffer.clear(); } } } else { if ( sendToErrorRow ) { putError( rowMeta, r, 1, errorMessage, null, "TOP001" ); outputRowData = null; } } return outputRowData; } TableOutput( StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta,
    Trans trans ); boolean processRow( StepMetaInterface smi, StepDataInterface sdi ); boolean isRowLevel(); boolean init( StepMetaInterface smi, StepDataInterface sdi ); void dispose( StepMetaInterface smi, StepDataInterface sdi ); }
TableOutput extends BaseStep implements StepInterface { protected Object[] writeToTable( RowMetaInterface rowMeta, Object[] r ) throws KettleException { if ( r == null ) { 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() ) { 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"; logError( 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() ) { 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 ) ) { 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() ) { insertRowData = new Object[data.valuenrs.length]; for ( int idx = 0; idx < data.valuenrs.length; idx++ ) { insertRowData[idx] = r[data.valuenrs[idx]]; } } if ( Utils.isEmpty( tableName ) ) { throw new KettleStepException( "The tablename is not defined (empty)" ); } insertStatement = data.preparedStatements.get( tableName ); if ( insertStatement == null ) { String sql = data.db .getInsertStatement( environmentSubstitute( meta.getSchemaName() ), tableName, data.insertRowMeta ); if ( log.isDetailed() ) { logDetailed( "Prepared statement : " + sql ); } if(meta.getDatabaseMeta().getURL().contains("phoenix")){ sql=sql.replace("insert", "upsert"); sql=sql.replace("INSERT", "UPSERT"); sql=sql.replace("update", "upsert"); sql=sql.replace("UPDATE", "UPSERT"); } try { insertStatement = data.db.prepareSQL( sql, meta.isReturningGeneratedKeys() ); } catch ( Exception e ) { logError( e.toString() ); } data.preparedStatements.put( tableName, insertStatement ); } try { if ( data.useSafePoints ) { data.savepoint = data.db.setSavepoint(); } data.db.setValues( data.insertRowMeta, insertRowData, insertStatement ); data.db.insertRow( insertStatement, data.batchMode, false ); if ( isRowLevel() ) { logRowlevel( "Written row: " + data.insertRowMeta.getString( insertRowData ) ); } Integer commitCounter = data.commitCounterMap.get( tableName ); if ( commitCounter == null ) { commitCounter = Integer.valueOf( 1 ); } else { commitCounter++; } data.commitCounterMap.put( tableName, Integer.valueOf( commitCounter.intValue() ) ); if ( data.useSafePoints ) { if ( data.releaseSavepoint ) { data.db.releaseSavepoint( data.savepoint ); } } if ( ( data.commitSize > 0 ) && ( ( commitCounter % data.commitSize ) == 0 ) ) { if ( data.db.getUseBatchInsert( data.batchMode ) ) { try { insertStatement.executeBatch(); data.db.commit(); insertStatement.clearBatch(); } catch ( SQLException ex ) { throw Database.createKettleDatabaseBatchException( "Error updating batch", ex ); } catch ( Exception ex ) { throw new KettleDatabaseException( "Unexpected error inserting row", ex ); } } else { data.db.commit(); } data.commitCounterMap.put( tableName, Integer.valueOf( 0 ) ); rowIsSafe = true; } else { rowIsSafe = false; } if ( meta.isReturningGeneratedKeys() ) { RowMetaAndData extraKeys = data.db.getGeneratedKeys( insertStatement ); if ( extraKeys.getRowMeta().size() > 0 ) { generatedKey = extraKeys.getRowMeta().getInteger( extraKeys.getData(), 0 ); } else { throw new KettleStepException( "No generated keys while \"return generated keys\" is active!" ); } } } catch ( KettleDatabaseBatchException be ) { errorMessage = be.toString(); batchProblem = true; sendToErrorRow = true; updateCounts = be.getUpdateCounts(); exceptionsList = be.getExceptionsList(); if ( getStepMeta().isDoingErrorHandling() ) { data.db.clearBatch( insertStatement ); data.db.commit( true ); } else { data.db.clearBatch( insertStatement ); data.db.rollback(); StringBuilder msg = new StringBuilder( "Error batch inserting rows into table [" + tableName + "]." ); msg.append( Const.CR ); msg.append( "Errors encountered (first 10):" ).append( Const.CR ); for ( int x = 0; x < be.getExceptionsList().size() && x < 10; x++ ) { Exception exception = be.getExceptionsList().get( x ); if ( exception.getMessage() != null ) { msg.append( exception.getMessage() ).append( Const.CR ); } } throw new KettleException( msg.toString(), be ); } } catch ( KettleDatabaseException dbe ) { if ( getStepMeta().isDoingErrorHandling() ) { if ( isRowLevel() ) { logRowlevel( "Written row to error handling : " + getInputRowMeta().getString( r ) ); } if ( data.useSafePoints ) { data.db.rollback( data.savepoint ); if ( data.releaseSavepoint ) { data.db.releaseSavepoint( data.savepoint ); } } sendToErrorRow = true; errorMessage = dbe.toString(); } else { if ( meta.ignoreErrors() ) { if ( data.warnings < 20 ) { if ( log.isBasic() ) { logBasic( "WARNING: Couldn't insert row into table: " + rowMeta.getString( r ) + Const.CR + dbe.getMessage() ); } } else if ( data.warnings == 20 ) { if ( log.isBasic() ) { logBasic( "FINAL WARNING (no more then 20 displayed): Couldn't insert row into table: " + rowMeta.getString( r ) + Const.CR + dbe.getMessage() ); } } data.warnings++; } else { setErrors( getErrors() + 1 ); data.db.rollback(); throw new KettleException( "Error inserting row into table [" + tableName + "] with values: " + rowMeta.getString( r ), dbe ); } } } if ( generatedKey != null ) { outputRowData = RowDataUtil.addValueData( outputRowData, rowMeta.size(), generatedKey ); } if ( data.batchMode ) { if ( sendToErrorRow ) { if ( batchProblem ) { data.batchBuffer.add( outputRowData ); outputRowData = null; processBatchException( errorMessage, updateCounts, exceptionsList ); } else { putError( rowMeta, r, 1L, errorMessage, null, "TOP001" ); outputRowData = null; } } else { data.batchBuffer.add( outputRowData ); outputRowData = null; if ( rowIsSafe ) { for ( int i = 0; i < data.batchBuffer.size(); i++ ) { Object[] row = data.batchBuffer.get( i ); putRow( data.outputRowMeta, row ); incrementLinesOutput(); } data.batchBuffer.clear(); } } } else { if ( sendToErrorRow ) { putError( rowMeta, r, 1, errorMessage, null, "TOP001" ); outputRowData = null; } } return outputRowData; } TableOutput( StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta,
    Trans trans ); boolean processRow( StepMetaInterface smi, StepDataInterface sdi ); boolean isRowLevel(); boolean init( StepMetaInterface smi, StepDataInterface sdi ); void dispose( StepMetaInterface smi, StepDataInterface sdi );  }