RecordInputFormat extends DelimitedInputFormat implements OutputSchemaProvider { @Override public void configure(Configuration config) { super.configure(config); numFields = config.getInteger(NUM_FIELDS_PARAMETER, -1); if (numFields < 1) { throw new IllegalArgumentException("Invalid configuration for RecordInputFormat: " + "Need to specify number of fields > 0."); } int[] textPosIdx = new int[numFields]; boolean anyTextPosSet = false; boolean allTextPosSet = true; int maxTextPos = -1; for (int i = 0; i < numFields; i++) { int pos = config.getInteger(TEXT_POSITION_PARAMETER_PREFIX + i, -1); if (pos == -1) { allTextPosSet = false; textPosIdx[i] = i; maxTextPos = i; } else { anyTextPosSet = true; textPosIdx[i] = pos; maxTextPos = pos > maxTextPos ? pos : maxTextPos; } } if(anyTextPosSet && !allTextPosSet) { throw new IllegalArgumentException("Invalid configuration for RecordInputFormat: " + "Not all text positions set"); } int[] recPosIdx = new int[numFields]; boolean anyRecPosSet = false; boolean allRecPosSet = true; for (int i = 0; i < numFields; i++) { int pos = config.getInteger(RECORD_POSITION_PARAMETER_PREFIX + i, -1); if (pos == -1) { allRecPosSet = false; recPosIdx[i] = i; } else { anyRecPosSet = true; recPosIdx[i] = pos; } } if(anyRecPosSet && !allRecPosSet) { throw new IllegalArgumentException("Invalid configuration for RecordInputFormat: " + "Not all record positions set"); } this.fieldParsers = new FieldParser[maxTextPos+1]; this.fieldValues = new Value[maxTextPos+1]; this.recordPositions = new int[maxTextPos+1]; for (int j = 0; j < maxTextPos; j++) { fieldParsers[j] = null; fieldValues[j] = null; recordPositions[j] = -1; } for (int i = 0; i < numFields; i++) { int pos = textPosIdx[i]; recordPositions[pos] = recPosIdx[i]; @SuppressWarnings("unchecked") Class<? extends FieldParser<Value>> clazz = (Class<? extends FieldParser<Value>>) config.getClass(FIELD_PARSER_PARAMETER_PREFIX + i, null); if (clazz == null) { throw new IllegalArgumentException("Invalid configuration for RecordInputFormat: " + "No field parser class for parameter " + i); } try { fieldParsers[pos] = clazz.newInstance(); } catch(InstantiationException ie) { throw new IllegalArgumentException("Invalid configuration for RecordInputFormat: " + "No field parser could not be instanciated for parameter " + i); } catch (IllegalAccessException e) { throw new IllegalArgumentException("Invalid configuration for RecordInputFormat: " + "No field parser could not be instanciated for parameter " + i); } fieldValues[pos] = fieldParsers[pos].getValue(); } final String fieldDelimStr = config.getString(FIELD_DELIMITER_PARAMETER, ","); if (fieldDelimStr.length() != 1) { throw new IllegalArgumentException("Invalid configuration for RecordInputFormat: " + "Field delimiter must be a single character"); } this.fieldDelim = fieldDelimStr.charAt(0); }  @Override void configure(Configuration config); @SuppressWarnings("unchecked") @Override boolean readRecord(PactRecord target, byte[] bytes, int offset, int numBytes); @Override int[] getOutputSchema(); static ConfigBuilder configureRecordFormat(FileDataSource target); static final String FIELD_DELIMITER_PARAMETER; static final String NUM_FIELDS_PARAMETER; static final String FIELD_PARSER_PARAMETER_PREFIX; static final String TEXT_POSITION_PARAMETER_PREFIX; static final String RECORD_POSITION_PARAMETER_PREFIX; static final String RECORD_DELIMITER_PARAMETER; }
@Test public void testConfigure() { try { Configuration config = new Configuration(); config.setString(RecordInputFormat.FILE_PARAMETER_KEY, "file: boolean validConfig = true; try { format.configure(config); } catch(IllegalArgumentException iae) { validConfig = false; } assertFalse(validConfig); config.setInteger(RecordInputFormat.NUM_FIELDS_PARAMETER, 2); validConfig = true; try { format.configure(config); } catch(IllegalArgumentException iae) { validConfig = false; } assertFalse(validConfig); config.setClass(RecordInputFormat.FIELD_PARSER_PARAMETER_PREFIX + 0, DecimalTextIntParser.class); config.setClass(RecordInputFormat.FIELD_PARSER_PARAMETER_PREFIX + 1, DecimalTextIntParser.class); validConfig = true; try { format.configure(config); } catch(IllegalArgumentException iae) { validConfig = false; } assertTrue(validConfig); config.setInteger(RecordInputFormat.NUM_FIELDS_PARAMETER, 3); validConfig = true; try { format.configure(config); } catch(IllegalArgumentException iae) { validConfig = false; } assertFalse(validConfig); config.setClass(RecordInputFormat.FIELD_PARSER_PARAMETER_PREFIX + 2, DecimalTextIntParser.class); config.setString(RecordInputFormat.FIELD_DELIMITER_PARAMETER, "||"); validConfig = true; try { format.configure(config); } catch(IllegalArgumentException iae) { validConfig = false; } assertFalse(validConfig); config.setString(RecordInputFormat.FIELD_DELIMITER_PARAMETER, "|"); validConfig = true; try { format.configure(config); } catch(IllegalArgumentException iae) { validConfig = false; } assertTrue(validConfig); config.setInteger(RecordInputFormat.TEXT_POSITION_PARAMETER_PREFIX + 1, 0); validConfig = true; try { format.configure(config); } catch(IllegalArgumentException iae) { validConfig = false; } assertFalse(validConfig); config.setInteger(RecordInputFormat.TEXT_POSITION_PARAMETER_PREFIX + 0, 3); config.setInteger(RecordInputFormat.TEXT_POSITION_PARAMETER_PREFIX + 2, 9); validConfig = true; try { format.configure(config); } catch(IllegalArgumentException iae) { validConfig = false; } assertTrue(validConfig); config.setInteger(RecordInputFormat.RECORD_POSITION_PARAMETER_PREFIX + 1, 0); validConfig = true; try { format.configure(config); } catch(IllegalArgumentException iae) { validConfig = false; } assertFalse(validConfig); config.setInteger(RecordInputFormat.RECORD_POSITION_PARAMETER_PREFIX + 0, 3); config.setInteger(RecordInputFormat.RECORD_POSITION_PARAMETER_PREFIX + 2, 9); validConfig = true; try { format.configure(config); } catch(IllegalArgumentException iae) { validConfig = false; } assertTrue(validConfig); } catch (Exception ex) { Assert.fail("Test failed due to a " + ex.getClass().getSimpleName() + ": " + ex.getMessage()); } }
@Test public void testReadNoPosAll() throws IOException { try { final String fileContent = "111|222|333|444|555\n666|777|888|999|000|"; final FileInputSplit split = createTempFile(fileContent); final Configuration parameters = new Configuration(); parameters.setString(RecordInputFormat.FILE_PARAMETER_KEY, "file: parameters.setInteger(RecordInputFormat.NUM_FIELDS_PARAMETER, 5); parameters.setString(RecordInputFormat.RECORD_DELIMITER_PARAMETER, "\n"); parameters.setString(RecordInputFormat.FIELD_DELIMITER_PARAMETER, "|"); parameters.setClass(RecordInputFormat.FIELD_PARSER_PARAMETER_PREFIX + 0, DecimalTextIntParser.class); parameters.setClass(RecordInputFormat.FIELD_PARSER_PARAMETER_PREFIX + 1, DecimalTextIntParser.class); parameters.setClass(RecordInputFormat.FIELD_PARSER_PARAMETER_PREFIX + 2, DecimalTextIntParser.class); parameters.setClass(RecordInputFormat.FIELD_PARSER_PARAMETER_PREFIX + 3, DecimalTextIntParser.class); parameters.setClass(RecordInputFormat.FIELD_PARSER_PARAMETER_PREFIX + 4, DecimalTextIntParser.class); format.configure(parameters); format.open(split); PactRecord record = new PactRecord(); assertTrue(format.nextRecord(record)); assertEquals(111, record.getField(0, PactInteger.class).getValue()); assertEquals(222, record.getField(1, PactInteger.class).getValue()); assertEquals(333, record.getField(2, PactInteger.class).getValue()); assertEquals(444, record.getField(3, PactInteger.class).getValue()); assertEquals(555, record.getField(4, PactInteger.class).getValue()); assertTrue(format.nextRecord(record)); assertEquals(666, record.getField(0, PactInteger.class).getValue()); assertEquals(777, record.getField(1, PactInteger.class).getValue()); assertEquals(888, record.getField(2, PactInteger.class).getValue()); assertEquals(999, record.getField(3, PactInteger.class).getValue()); assertEquals(000, record.getField(4, PactInteger.class).getValue()); assertFalse(format.nextRecord(record)); assertTrue(format.reachedEnd()); } catch (Exception ex) { Assert.fail("Test failed due to a " + ex.getClass().getSimpleName() + ": " + ex.getMessage()); } }
@Test public void testReadNoPosFirstN() throws IOException { try { final String fileContent = "111|222|333|444|555|\n666|777|888|999|000|"; final FileInputSplit split = createTempFile(fileContent); final Configuration parameters = new Configuration(); parameters.setString(RecordInputFormat.FILE_PARAMETER_KEY, "file: parameters.setInteger(RecordInputFormat.NUM_FIELDS_PARAMETER, 2); parameters.setString(RecordInputFormat.RECORD_DELIMITER_PARAMETER, "\n"); parameters.setString(RecordInputFormat.FIELD_DELIMITER_PARAMETER, "|"); parameters.setClass(RecordInputFormat.FIELD_PARSER_PARAMETER_PREFIX + 0, DecimalTextIntParser.class); parameters.setClass(RecordInputFormat.FIELD_PARSER_PARAMETER_PREFIX + 1, DecimalTextIntParser.class); format.configure(parameters); format.open(split); PactRecord record = new PactRecord(); assertTrue(format.nextRecord(record)); assertEquals(111, record.getField(0, PactInteger.class).getValue()); assertEquals(222, record.getField(1, PactInteger.class).getValue()); boolean notParsed = false; try { record.getField(2, PactInteger.class); } catch (IndexOutOfBoundsException ioo) { notParsed = true; } assertTrue(notParsed); assertTrue(format.nextRecord(record)); assertEquals(666, record.getField(0, PactInteger.class).getValue()); assertEquals(777, record.getField(1, PactInteger.class).getValue()); notParsed = false; try { record.getField(2, PactInteger.class); } catch (IndexOutOfBoundsException ioo) { notParsed = true; } assertTrue(notParsed); assertFalse(format.nextRecord(record)); assertTrue(format.reachedEnd()); } catch (Exception ex) { Assert.fail("Test failed due to a " + ex.getClass().getSimpleName() + ": " + ex.getMessage()); } }
@Test public void testReadTextPos() throws IOException { try { final String fileContent = "111|222|333|444|555|666|777|888|999|000|\n000|999|888|777|666|555|444|333|222|111|"; final FileInputSplit split = createTempFile(fileContent); final Configuration parameters = new Configuration(); parameters.setString(RecordInputFormat.FILE_PARAMETER_KEY, "file: parameters.setInteger(RecordInputFormat.NUM_FIELDS_PARAMETER, 3); parameters.setString(RecordInputFormat.RECORD_DELIMITER_PARAMETER, "\n"); parameters.setString(RecordInputFormat.FIELD_DELIMITER_PARAMETER, "|"); parameters.setClass(RecordInputFormat.FIELD_PARSER_PARAMETER_PREFIX + 0, DecimalTextIntParser.class); parameters.setClass(RecordInputFormat.FIELD_PARSER_PARAMETER_PREFIX + 1, DecimalTextIntParser.class); parameters.setClass(RecordInputFormat.FIELD_PARSER_PARAMETER_PREFIX + 2, DecimalTextIntParser.class); parameters.setInteger(RecordInputFormat.TEXT_POSITION_PARAMETER_PREFIX + 0, 0); parameters.setInteger(RecordInputFormat.TEXT_POSITION_PARAMETER_PREFIX + 1, 3); parameters.setInteger(RecordInputFormat.TEXT_POSITION_PARAMETER_PREFIX + 2, 7); format.configure(parameters); format.open(split); PactRecord record = new PactRecord(); assertTrue(format.nextRecord(record)); assertEquals(111, record.getField(0, PactInteger.class).getValue()); assertEquals(444, record.getField(1, PactInteger.class).getValue()); assertEquals(888, record.getField(2, PactInteger.class).getValue()); assertTrue(format.nextRecord(record)); assertEquals(000, record.getField(0, PactInteger.class).getValue()); assertEquals(777, record.getField(1, PactInteger.class).getValue()); assertEquals(333, record.getField(2, PactInteger.class).getValue()); assertFalse(format.nextRecord(record)); assertTrue(format.reachedEnd()); parameters.setInteger(RecordInputFormat.TEXT_POSITION_PARAMETER_PREFIX + 0, 8); parameters.setInteger(RecordInputFormat.TEXT_POSITION_PARAMETER_PREFIX + 1, 1); parameters.setInteger(RecordInputFormat.TEXT_POSITION_PARAMETER_PREFIX + 2, 3); format.configure(parameters); format.open(split); assertTrue(format.nextRecord(record)); assertEquals(999, record.getField(0, PactInteger.class).getValue()); assertEquals(222, record.getField(1, PactInteger.class).getValue()); assertEquals(444, record.getField(2, PactInteger.class).getValue()); assertTrue(format.nextRecord(record)); assertEquals(222, record.getField(0, PactInteger.class).getValue()); assertEquals(999, record.getField(1, PactInteger.class).getValue()); assertEquals(777, record.getField(2, PactInteger.class).getValue()); assertFalse(format.nextRecord(record)); assertTrue(format.reachedEnd()); } catch (Exception ex) { Assert.fail("Test failed due to a " + ex.getClass().getSimpleName() + ": " + ex.getMessage()); } }
@Test public void testReadRecPos() throws IOException { try { final String fileContent = "111|222|333|\n444|555|666|"; final FileInputSplit split = createTempFile(fileContent); final Configuration parameters = new Configuration(); parameters.setString(RecordInputFormat.FILE_PARAMETER_KEY, "file: parameters.setInteger(RecordInputFormat.NUM_FIELDS_PARAMETER, 3); parameters.setString(RecordInputFormat.RECORD_DELIMITER_PARAMETER, "\n"); parameters.setString(RecordInputFormat.FIELD_DELIMITER_PARAMETER, "|"); parameters.setClass(RecordInputFormat.FIELD_PARSER_PARAMETER_PREFIX + 0, DecimalTextIntParser.class); parameters.setClass(RecordInputFormat.FIELD_PARSER_PARAMETER_PREFIX + 1, DecimalTextIntParser.class); parameters.setClass(RecordInputFormat.FIELD_PARSER_PARAMETER_PREFIX + 2, DecimalTextIntParser.class); parameters.setInteger(RecordInputFormat.RECORD_POSITION_PARAMETER_PREFIX + 0, 1); parameters.setInteger(RecordInputFormat.RECORD_POSITION_PARAMETER_PREFIX + 1, 2); parameters.setInteger(RecordInputFormat.RECORD_POSITION_PARAMETER_PREFIX + 2, 0); format.configure(parameters); format.open(split); PactRecord record = new PactRecord(); assertTrue(format.nextRecord(record)); assertEquals(111, record.getField(1, PactInteger.class).getValue()); assertEquals(222, record.getField(2, PactInteger.class).getValue()); assertEquals(333, record.getField(0, PactInteger.class).getValue()); assertTrue(format.nextRecord(record)); assertEquals(444, record.getField(1, PactInteger.class).getValue()); assertEquals(555, record.getField(2, PactInteger.class).getValue()); assertEquals(666, record.getField(0, PactInteger.class).getValue()); assertFalse(format.nextRecord(record)); assertTrue(format.reachedEnd()); parameters.setInteger(RecordInputFormat.RECORD_POSITION_PARAMETER_PREFIX + 0, 1); parameters.setInteger(RecordInputFormat.RECORD_POSITION_PARAMETER_PREFIX + 1, 8); parameters.setInteger(RecordInputFormat.RECORD_POSITION_PARAMETER_PREFIX + 2, 3); format.configure(parameters); format.open(split); assertTrue(format.nextRecord(record)); assertEquals(111, record.getField(1, PactInteger.class).getValue()); assertEquals(222, record.getField(8, PactInteger.class).getValue()); assertEquals(333, record.getField(3, PactInteger.class).getValue()); assertTrue(format.nextRecord(record)); assertEquals(444, record.getField(1, PactInteger.class).getValue()); assertEquals(555, record.getField(8, PactInteger.class).getValue()); assertEquals(666, record.getField(3, PactInteger.class).getValue()); assertFalse(format.nextRecord(record)); assertTrue(format.reachedEnd()); } catch (Exception ex) { Assert.fail("Test failed due to a " + ex.getClass().getSimpleName() + ": " + ex.getMessage()); } }
@Test public void testReadTextPosRecPos() throws IOException { try { final String fileContent = "111|222|333|444|555|666|777|888|999|000|\n000|999|888|777|666|555|444|333|222|111|"; final FileInputSplit split = createTempFile(fileContent); final Configuration parameters = new Configuration(); parameters.setString(RecordInputFormat.FILE_PARAMETER_KEY, "file: parameters.setInteger(RecordInputFormat.NUM_FIELDS_PARAMETER, 5); parameters.setString(RecordInputFormat.RECORD_DELIMITER_PARAMETER, "\n"); parameters.setString(RecordInputFormat.FIELD_DELIMITER_PARAMETER, "|"); parameters.setClass(RecordInputFormat.FIELD_PARSER_PARAMETER_PREFIX + 0, DecimalTextIntParser.class); parameters.setClass(RecordInputFormat.FIELD_PARSER_PARAMETER_PREFIX + 1, DecimalTextIntParser.class); parameters.setClass(RecordInputFormat.FIELD_PARSER_PARAMETER_PREFIX + 2, DecimalTextIntParser.class); parameters.setClass(RecordInputFormat.FIELD_PARSER_PARAMETER_PREFIX + 3, DecimalTextIntParser.class); parameters.setClass(RecordInputFormat.FIELD_PARSER_PARAMETER_PREFIX + 4, DecimalTextIntParser.class); parameters.setInteger(RecordInputFormat.TEXT_POSITION_PARAMETER_PREFIX + 0, 3); parameters.setInteger(RecordInputFormat.TEXT_POSITION_PARAMETER_PREFIX + 1, 0); parameters.setInteger(RecordInputFormat.TEXT_POSITION_PARAMETER_PREFIX + 2, 1); parameters.setInteger(RecordInputFormat.TEXT_POSITION_PARAMETER_PREFIX + 3, 8); parameters.setInteger(RecordInputFormat.TEXT_POSITION_PARAMETER_PREFIX + 4, 4); parameters.setInteger(RecordInputFormat.RECORD_POSITION_PARAMETER_PREFIX + 0, 2); parameters.setInteger(RecordInputFormat.RECORD_POSITION_PARAMETER_PREFIX + 1, 0); parameters.setInteger(RecordInputFormat.RECORD_POSITION_PARAMETER_PREFIX + 2, 1); parameters.setInteger(RecordInputFormat.RECORD_POSITION_PARAMETER_PREFIX + 3, 3); parameters.setInteger(RecordInputFormat.RECORD_POSITION_PARAMETER_PREFIX + 4, 6); format.configure(parameters); format.open(split); PactRecord record = new PactRecord(); assertTrue(format.nextRecord(record)); assertEquals(444, record.getField(2, PactInteger.class).getValue()); assertEquals(111, record.getField(0, PactInteger.class).getValue()); assertEquals(222, record.getField(1, PactInteger.class).getValue()); assertEquals(999, record.getField(3, PactInteger.class).getValue()); assertEquals(555, record.getField(6, PactInteger.class).getValue()); assertTrue(format.nextRecord(record)); assertEquals(777, record.getField(2, PactInteger.class).getValue()); assertEquals(000, record.getField(0, PactInteger.class).getValue()); assertEquals(999, record.getField(1, PactInteger.class).getValue()); assertEquals(222, record.getField(3, PactInteger.class).getValue()); assertEquals(666, record.getField(6, PactInteger.class).getValue()); assertFalse(format.nextRecord(record)); assertTrue(format.reachedEnd()); } catch (Exception ex) { Assert.fail("Test failed due to a " + ex.getClass().getSimpleName() + ": " + ex.getMessage()); } }
@Test public void testReadTooShortInput() throws IOException { try { final String fileContent = "111|222|333|444\n666|777|888|999"; final FileInputSplit split = createTempFile(fileContent); final Configuration parameters = new Configuration(); parameters.setString(RecordInputFormat.FILE_PARAMETER_KEY, "file: parameters.setInteger(RecordInputFormat.NUM_FIELDS_PARAMETER, 5); parameters.setString(RecordInputFormat.RECORD_DELIMITER_PARAMETER, "\n"); parameters.setString(RecordInputFormat.FIELD_DELIMITER_PARAMETER, "|"); parameters.setClass(RecordInputFormat.FIELD_PARSER_PARAMETER_PREFIX + 0, DecimalTextIntParser.class); parameters.setClass(RecordInputFormat.FIELD_PARSER_PARAMETER_PREFIX + 1, DecimalTextIntParser.class); parameters.setClass(RecordInputFormat.FIELD_PARSER_PARAMETER_PREFIX + 2, DecimalTextIntParser.class); parameters.setClass(RecordInputFormat.FIELD_PARSER_PARAMETER_PREFIX + 3, DecimalTextIntParser.class); parameters.setClass(RecordInputFormat.FIELD_PARSER_PARAMETER_PREFIX + 4, DecimalTextIntParser.class); format.configure(parameters); format.open(split); PactRecord record = new PactRecord(); assertFalse(format.nextRecord(record)); } catch (Exception ex) { Assert.fail("Test failed due to a " + ex.getClass().getSimpleName() + ": " + ex.getMessage()); } }