@Test public void testProcessRow() throws Exception { boolean result; int rowsProcessed = 0; ArgumentCaptor<RowMeta> rowMetaCaptor = ArgumentCaptor.forClass( RowMeta.class ); ArgumentCaptor<Object[]> dataCaptor = ArgumentCaptor.forClass( Object[].class ); do { result = orcInput.processRow( orcInputMeta, orcInputData ); if ( result ) { rowsProcessed++; } } while ( result ); assertEquals( 2, rowsProcessed ); verify( mockRowHandler, times( 2 ) ).putRow( rowMetaCaptor.capture(), dataCaptor.capture() ); List<RowMeta> rowMeta = rowMetaCaptor.getAllValues(); List<Object[]> dataCaptured = dataCaptor.getAllValues(); for ( int rowNum = 0; rowNum < 2; rowNum++ ) { assertEquals( 0, rowMeta.get( rowNum ).indexOfValue( "str" ) ); assertEquals( "string" + ( rowNum % 2 + 1 ), dataCaptured.get( rowNum )[ 0 ] ); } }
@Override public boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException { meta = (OrcInputMeta) smi; data = (OrcInputData) sdi; try { if ( data.input == null || data.reader == null || data.rowIterator == null ) { FormatService formatService = getFormatService(); if ( meta.inputFiles == null || meta.getFilename() == null || meta.getFilename().length() == 0 ) { throw new KettleException( "No input files defined" ); } data.input = formatService.createInputFormat( IPentahoOrcInputFormat.class, getNamedCluster() ); String inputFileName = getKettleVFSFileName( meta.getParentStepMeta().getParentTransMeta().environmentSubstitute( meta.getFilename() ) ); data.input.setInputFile( inputFileName ); data.input.setSchema( createSchemaFromMeta( meta ) ); data.reader = data.input.createRecordReader( null ); data.rowIterator = data.reader.iterator(); } if ( data.rowIterator.hasNext() ) { RowMetaAndData row = data.rowIterator.next(); putRow( row.getRowMeta(), row.getData() ); return true; } else { data.reader.close(); data.reader = null; data.input = null; setOutputDone(); return false; } } catch ( KettleException ex ) { throw ex; } catch ( Exception ex ) { throw new KettleException( ex ); } }
OrcInput extends BaseFileInputStep<OrcInputMeta, OrcInputData> { @Override public boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException { meta = (OrcInputMeta) smi; data = (OrcInputData) sdi; try { if ( data.input == null || data.reader == null || data.rowIterator == null ) { FormatService formatService = getFormatService(); if ( meta.inputFiles == null || meta.getFilename() == null || meta.getFilename().length() == 0 ) { throw new KettleException( "No input files defined" ); } data.input = formatService.createInputFormat( IPentahoOrcInputFormat.class, getNamedCluster() ); String inputFileName = getKettleVFSFileName( meta.getParentStepMeta().getParentTransMeta().environmentSubstitute( meta.getFilename() ) ); data.input.setInputFile( inputFileName ); data.input.setSchema( createSchemaFromMeta( meta ) ); data.reader = data.input.createRecordReader( null ); data.rowIterator = data.reader.iterator(); } if ( data.rowIterator.hasNext() ) { RowMetaAndData row = data.rowIterator.next(); putRow( row.getRowMeta(), row.getData() ); return true; } else { data.reader.close(); data.reader = null; data.input = null; setOutputDone(); return false; } } catch ( KettleException ex ) { throw ex; } catch ( Exception ex ) { throw new KettleException( ex ); } } }
OrcInput extends BaseFileInputStep<OrcInputMeta, OrcInputData> { @Override public boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException { meta = (OrcInputMeta) smi; data = (OrcInputData) sdi; try { if ( data.input == null || data.reader == null || data.rowIterator == null ) { FormatService formatService = getFormatService(); if ( meta.inputFiles == null || meta.getFilename() == null || meta.getFilename().length() == 0 ) { throw new KettleException( "No input files defined" ); } data.input = formatService.createInputFormat( IPentahoOrcInputFormat.class, getNamedCluster() ); String inputFileName = getKettleVFSFileName( meta.getParentStepMeta().getParentTransMeta().environmentSubstitute( meta.getFilename() ) ); data.input.setInputFile( inputFileName ); data.input.setSchema( createSchemaFromMeta( meta ) ); data.reader = data.input.createRecordReader( null ); data.rowIterator = data.reader.iterator(); } if ( data.rowIterator.hasNext() ) { RowMetaAndData row = data.rowIterator.next(); putRow( row.getRowMeta(), row.getData() ); return true; } else { data.reader.close(); data.reader = null; data.input = null; setOutputDone(); return false; } } catch ( KettleException ex ) { throw ex; } catch ( Exception ex ) { throw new KettleException( ex ); } } OrcInput( StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta,
                   Trans trans ); }
OrcInput extends BaseFileInputStep<OrcInputMeta, OrcInputData> { @Override public boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException { meta = (OrcInputMeta) smi; data = (OrcInputData) sdi; try { if ( data.input == null || data.reader == null || data.rowIterator == null ) { FormatService formatService = getFormatService(); if ( meta.inputFiles == null || meta.getFilename() == null || meta.getFilename().length() == 0 ) { throw new KettleException( "No input files defined" ); } data.input = formatService.createInputFormat( IPentahoOrcInputFormat.class, getNamedCluster() ); String inputFileName = getKettleVFSFileName( meta.getParentStepMeta().getParentTransMeta().environmentSubstitute( meta.getFilename() ) ); data.input.setInputFile( inputFileName ); data.input.setSchema( createSchemaFromMeta( meta ) ); data.reader = data.input.createRecordReader( null ); data.rowIterator = data.reader.iterator(); } if ( data.rowIterator.hasNext() ) { RowMetaAndData row = data.rowIterator.next(); putRow( row.getRowMeta(), row.getData() ); return true; } else { data.reader.close(); data.reader = null; data.input = null; setOutputDone(); return false; } } catch ( KettleException ex ) { throw ex; } catch ( Exception ex ) { throw new KettleException( ex ); } } OrcInput( StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta,
                   Trans trans ); @Override boolean processRow( StepMetaInterface smi, StepDataInterface sdi ); static List<IOrcInputField> retrieveSchema( NamedClusterServiceLocator namedClusterServiceLocator,
                                                     NamedCluster namedCluster, String dataPath ); static List<IOrcInputField> createSchemaFromMeta( OrcInputMetaBase meta ); static String getKettleVFSFileName( String path ); }
OrcInput extends BaseFileInputStep<OrcInputMeta, OrcInputData> { @Override public boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException { meta = (OrcInputMeta) smi; data = (OrcInputData) sdi; try { if ( data.input == null || data.reader == null || data.rowIterator == null ) { FormatService formatService = getFormatService(); if ( meta.inputFiles == null || meta.getFilename() == null || meta.getFilename().length() == 0 ) { throw new KettleException( "No input files defined" ); } data.input = formatService.createInputFormat( IPentahoOrcInputFormat.class, getNamedCluster() ); String inputFileName = getKettleVFSFileName( meta.getParentStepMeta().getParentTransMeta().environmentSubstitute( meta.getFilename() ) ); data.input.setInputFile( inputFileName ); data.input.setSchema( createSchemaFromMeta( meta ) ); data.reader = data.input.createRecordReader( null ); data.rowIterator = data.reader.iterator(); } if ( data.rowIterator.hasNext() ) { RowMetaAndData row = data.rowIterator.next(); putRow( row.getRowMeta(), row.getData() ); return true; } else { data.reader.close(); data.reader = null; data.input = null; setOutputDone(); return false; } } catch ( KettleException ex ) { throw ex; } catch ( Exception ex ) { throw new KettleException( ex ); } } OrcInput( StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta,
                   Trans trans ); @Override boolean processRow( StepMetaInterface smi, StepDataInterface sdi ); static List<IOrcInputField> retrieveSchema( NamedClusterServiceLocator namedClusterServiceLocator,
                                                     NamedCluster namedCluster, String dataPath ); static List<IOrcInputField> createSchemaFromMeta( OrcInputMetaBase meta ); static String getKettleVFSFileName( String path ); static final long SPLIT_SIZE; }