ParquetInput extends BaseFileInputStep<ParquetInputMeta, ParquetInputData> { @Override public boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException { meta = (ParquetInputMeta) smi; data = (ParquetInputData) sdi; try { if ( data.splits == null ) { initSplits(); } if ( data.currentSplit >= data.splits.size() ) { setOutputDone(); return false; } if ( data.reader == null ) { openReader( data ); } if ( data.rowIterator.hasNext() ) { RowMetaAndData row = data.rowIterator.next(); putRow( row.getRowMeta(), row.getData() ); return true; } else { data.reader.close(); data.reader = null; logDebug( "Close split {0}", data.currentSplit ); data.currentSplit++; return true; } } catch ( NoSuchFileException ex ) { throw new KettleException( "No input file" ); } catch ( KettleException ex ) { throw ex; } catch ( Exception ex ) { throw new KettleException( ex ); } } ParquetInput( StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta,
                       Trans trans ); static List<? extends IParquetInputField> retrieveSchema(
    NamedClusterServiceLocator namedClusterServiceLocator, NamedCluster namedCluster, String path ); static List<IParquetInputField> createSchemaFromMeta( ParquetInputMetaBase meta ); @Override boolean processRow( StepMetaInterface smi, StepDataInterface sdi ); static final long SPLIT_SIZE; }
@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 = parquetInput.processRow( parquetInputMeta, parquetInputData ); if ( result ) { rowsProcessed++; } } while ( result ); assertEquals( 3, 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 ] ); } }