@Test public void testMissingField() throws IOException, MalformedRecordException { final List<RecordField> fields = getDefaultFields(); fields.replaceAll(f -> f.getFieldName().equals("balance") ? new RecordField("balance", doubleDataType) : f); final RecordSchema schema = new SimpleRecordSchema(fields); final String headerLine = "id, name, balance, address, city, state, zipCode, country"; final String inputRecord = "1, John, 40.80, 123 My Street, My City, MS, 11111"; final String csvData = headerLine + "\n" + inputRecord; final byte[] inputData = csvData.getBytes(); try (final InputStream bais = new ByteArrayInputStream(inputData); final CSVRecordReader reader = createReader(bais, schema, format)) { final Record record = reader.nextRecord(); assertNotNull(record); assertEquals("1", record.getValue("id")); assertEquals("John", record.getValue("name")); assertEquals(40.8D, record.getValue("balance")); assertEquals("123 My Street", record.getValue("address")); assertEquals("My City", record.getValue("city")); assertEquals("MS", record.getValue("state")); assertEquals("11111", record.getValue("zipCode")); assertNull(record.getValue("country")); assertNull(reader.nextRecord()); } }
@Override public Record nextRecord(final boolean coerceTypes, final boolean dropUnknownFields) throws IOException, MalformedRecordException { final RecordSchema schema = getSchema(); final List<RecordField> recordFields = getRecordFields(); final int numFieldNames = recordFields.size(); for (final CSVRecord csvRecord : csvParser) { final Map<String, Object> values = new LinkedHashMap<>(recordFields.size() * 2); for (int i = 0; i < csvRecord.size(); i++) { final String rawValue = csvRecord.get(i); final String rawFieldName; final DataType dataType; if (i >= numFieldNames) { if (!dropUnknownFields) { values.put("unknown_field_index_" + i, rawValue); } continue; } else { final RecordField recordField = recordFields.get(i); rawFieldName = recordField.getFieldName(); dataType = recordField.getDataType(); } final Object value; if (coerceTypes) { value = convert(rawValue, dataType, rawFieldName); } else { value = convertSimpleIfPossible(rawValue, dataType, rawFieldName); } values.put(rawFieldName, value); } return new MapRecord(schema, values, coerceTypes, dropUnknownFields); } return null; }
CSVRecordReader extends AbstractCSVRecordReader { @Override public Record nextRecord(final boolean coerceTypes, final boolean dropUnknownFields) throws IOException, MalformedRecordException { final RecordSchema schema = getSchema(); final List<RecordField> recordFields = getRecordFields(); final int numFieldNames = recordFields.size(); for (final CSVRecord csvRecord : csvParser) { final Map<String, Object> values = new LinkedHashMap<>(recordFields.size() * 2); for (int i = 0; i < csvRecord.size(); i++) { final String rawValue = csvRecord.get(i); final String rawFieldName; final DataType dataType; if (i >= numFieldNames) { if (!dropUnknownFields) { values.put("unknown_field_index_" + i, rawValue); } continue; } else { final RecordField recordField = recordFields.get(i); rawFieldName = recordField.getFieldName(); dataType = recordField.getDataType(); } final Object value; if (coerceTypes) { value = convert(rawValue, dataType, rawFieldName); } else { value = convertSimpleIfPossible(rawValue, dataType, rawFieldName); } values.put(rawFieldName, value); } return new MapRecord(schema, values, coerceTypes, dropUnknownFields); } return null; } }
CSVRecordReader extends AbstractCSVRecordReader { @Override public Record nextRecord(final boolean coerceTypes, final boolean dropUnknownFields) throws IOException, MalformedRecordException { final RecordSchema schema = getSchema(); final List<RecordField> recordFields = getRecordFields(); final int numFieldNames = recordFields.size(); for (final CSVRecord csvRecord : csvParser) { final Map<String, Object> values = new LinkedHashMap<>(recordFields.size() * 2); for (int i = 0; i < csvRecord.size(); i++) { final String rawValue = csvRecord.get(i); final String rawFieldName; final DataType dataType; if (i >= numFieldNames) { if (!dropUnknownFields) { values.put("unknown_field_index_" + i, rawValue); } continue; } else { final RecordField recordField = recordFields.get(i); rawFieldName = recordField.getFieldName(); dataType = recordField.getDataType(); } final Object value; if (coerceTypes) { value = convert(rawValue, dataType, rawFieldName); } else { value = convertSimpleIfPossible(rawValue, dataType, rawFieldName); } values.put(rawFieldName, value); } return new MapRecord(schema, values, coerceTypes, dropUnknownFields); } return null; } CSVRecordReader(final InputStream in, final ComponentLog logger, final RecordSchema schema, final CSVFormat csvFormat, final boolean hasHeader, final boolean ignoreHeader,
                           final String dateFormat, final String timeFormat, final String timestampFormat, final String encoding); }
CSVRecordReader extends AbstractCSVRecordReader { @Override public Record nextRecord(final boolean coerceTypes, final boolean dropUnknownFields) throws IOException, MalformedRecordException { final RecordSchema schema = getSchema(); final List<RecordField> recordFields = getRecordFields(); final int numFieldNames = recordFields.size(); for (final CSVRecord csvRecord : csvParser) { final Map<String, Object> values = new LinkedHashMap<>(recordFields.size() * 2); for (int i = 0; i < csvRecord.size(); i++) { final String rawValue = csvRecord.get(i); final String rawFieldName; final DataType dataType; if (i >= numFieldNames) { if (!dropUnknownFields) { values.put("unknown_field_index_" + i, rawValue); } continue; } else { final RecordField recordField = recordFields.get(i); rawFieldName = recordField.getFieldName(); dataType = recordField.getDataType(); } final Object value; if (coerceTypes) { value = convert(rawValue, dataType, rawFieldName); } else { value = convertSimpleIfPossible(rawValue, dataType, rawFieldName); } values.put(rawFieldName, value); } return new MapRecord(schema, values, coerceTypes, dropUnknownFields); } return null; } CSVRecordReader(final InputStream in, final ComponentLog logger, final RecordSchema schema, final CSVFormat csvFormat, final boolean hasHeader, final boolean ignoreHeader,
                           final String dateFormat, final String timeFormat, final String timestampFormat, final String encoding); @Override Record nextRecord(final boolean coerceTypes, final boolean dropUnknownFields); @Override void close(); }
CSVRecordReader extends AbstractCSVRecordReader { @Override public Record nextRecord(final boolean coerceTypes, final boolean dropUnknownFields) throws IOException, MalformedRecordException { final RecordSchema schema = getSchema(); final List<RecordField> recordFields = getRecordFields(); final int numFieldNames = recordFields.size(); for (final CSVRecord csvRecord : csvParser) { final Map<String, Object> values = new LinkedHashMap<>(recordFields.size() * 2); for (int i = 0; i < csvRecord.size(); i++) { final String rawValue = csvRecord.get(i); final String rawFieldName; final DataType dataType; if (i >= numFieldNames) { if (!dropUnknownFields) { values.put("unknown_field_index_" + i, rawValue); } continue; } else { final RecordField recordField = recordFields.get(i); rawFieldName = recordField.getFieldName(); dataType = recordField.getDataType(); } final Object value; if (coerceTypes) { value = convert(rawValue, dataType, rawFieldName); } else { value = convertSimpleIfPossible(rawValue, dataType, rawFieldName); } values.put(rawFieldName, value); } return new MapRecord(schema, values, coerceTypes, dropUnknownFields); } return null; } CSVRecordReader(final InputStream in, final ComponentLog logger, final RecordSchema schema, final CSVFormat csvFormat, final boolean hasHeader, final boolean ignoreHeader,
                           final String dateFormat, final String timeFormat, final String timestampFormat, final String encoding); @Override Record nextRecord(final boolean coerceTypes, final boolean dropUnknownFields); @Override void close();  }