@Test public void testJoinRowsStep() throws Exception { JoinRowsMeta joinRowsMeta = new JoinRowsMeta(); joinRowsMeta.setMainStepname( "main step name" ); joinRowsMeta.setPrefix( "out" ); joinRowsMeta.setCacheSize( 3 ); JoinRowsData joinRowsData = new JoinRowsData(); JoinRows joinRows = getJoinRows(); joinRows.getTrans().setRunning( true ); joinRows.init( joinRowsMeta, joinRowsData ); List<RowSet> rowSets = new ArrayList<>(); rowSets.add( getRowSetWithData( 3, "main --", true ) ); rowSets.add( getRowSetWithData( 3, "secondary --", false ) ); joinRows.setInputRowSets( rowSets ); RowStepCollector rowStepCollector = new RowStepCollector(); joinRows.addRowListener( rowStepCollector ); joinRows.getLogChannel().setLogLevel( LogLevel.ROWLEVEL ); KettleLogStore.init(); while ( true ) { if ( !joinRows.processRow( joinRowsMeta, joinRowsData ) ) { break; } } rowStepCollector.getRowsWritten(); assertEquals( 9, rowStepCollector.getRowsWritten().size() ); assertEquals( 6, rowStepCollector.getRowsRead().size() ); Object[][] expectedResult = createExpectedResult(); List<Object[]> rowWritten = rowStepCollector.getRowsWritten().stream().map( RowMetaAndData::getData ).collect( Collectors.toList() ); for ( int i = 0; i < 9; i++ ) { assertTrue( Arrays.equals( expectedResult[i], rowWritten.get( i ) ) ); } }
public boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException { meta = (JoinRowsMeta) smi; data = (JoinRowsData) sdi; if ( first ) { first = false; initialize(); } if ( data.caching ) { if ( !cacheInputRow() ) { return false; } } else { if ( !outputRow() ) { return false; } } return true; }
JoinRows extends BaseStep implements StepInterface { public boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException { meta = (JoinRowsMeta) smi; data = (JoinRowsData) sdi; if ( first ) { first = false; initialize(); } if ( data.caching ) { if ( !cacheInputRow() ) { return false; } } else { if ( !outputRow() ) { return false; } } return true; } }
JoinRows extends BaseStep implements StepInterface { public boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException { meta = (JoinRowsMeta) smi; data = (JoinRowsData) sdi; if ( first ) { first = false; initialize(); } if ( data.caching ) { if ( !cacheInputRow() ) { return false; } } else { if ( !outputRow() ) { return false; } } return true; } JoinRows( StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta,
    Trans trans ); }
JoinRows extends BaseStep implements StepInterface { public boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException { meta = (JoinRowsMeta) smi; data = (JoinRowsData) sdi; if ( first ) { first = false; initialize(); } if ( data.caching ) { if ( !cacheInputRow() ) { return false; } } else { if ( !outputRow() ) { return false; } } return true; } JoinRows( StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta,
    Trans trans ); @SuppressWarnings( "unchecked" ) void initialize(); Object[] getRowData( int filenr ); boolean processRow( StepMetaInterface smi, StepDataInterface sdi ); void dispose( StepMetaInterface smi, StepDataInterface sdi ); @Override void batchComplete(); }
JoinRows extends BaseStep implements StepInterface { public boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException { meta = (JoinRowsMeta) smi; data = (JoinRowsData) sdi; if ( first ) { first = false; initialize(); } if ( data.caching ) { if ( !cacheInputRow() ) { return false; } } else { if ( !outputRow() ) { return false; } } return true; } JoinRows( StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta,
    Trans trans ); @SuppressWarnings( "unchecked" ) void initialize(); Object[] getRowData( int filenr ); boolean processRow( StepMetaInterface smi, StepDataInterface sdi ); void dispose( StepMetaInterface smi, StepDataInterface sdi ); @Override void batchComplete();  }