@Test public void testSimple() throws SchemaNotFoundException, IOException { final String headerLine = "a, b, c, d, e\\,z, f"; final byte[] headerBytes = headerLine.getBytes(); final Map<PropertyDescriptor, String> properties = new HashMap<>(); properties.put(CSVUtils.CSV_FORMAT, CSVUtils.CUSTOM.getValue()); properties.put(CSVUtils.COMMENT_MARKER, "#"); properties.put(CSVUtils.VALUE_SEPARATOR, ","); properties.put(CSVUtils.TRIM_FIELDS, "true"); properties.put(CSVUtils.QUOTE_CHAR, "\""); properties.put(CSVUtils.ESCAPE_CHAR, "\\"); final ConfigurationContext context = new MockConfigurationContext(properties, null); final CSVHeaderSchemaStrategy strategy = new CSVHeaderSchemaStrategy(context); final RecordSchema schema; try (final InputStream bais = new ByteArrayInputStream(headerBytes)) { schema = strategy.getSchema(null, bais, null); } final List<String> expectedFieldNames = Arrays.asList("a", "b", "c", "d", "e,z", "f"); assertEquals(expectedFieldNames, schema.getFieldNames()); assertTrue(schema.getFields().stream() .allMatch(field -> field.getDataType().equals(RecordFieldType.STRING.getDataType()))); }
@Override public RecordSchema getSchema(Map<String, String> variables, final InputStream contentStream, final RecordSchema readSchema) throws SchemaNotFoundException { if (this.context == null) { throw new SchemaNotFoundException("Schema Access Strategy intended only for validation purposes and cannot obtain schema"); } try { final CSVFormat csvFormat = CSVUtils.createCSVFormat(context).withFirstRecordAsHeader(); try (final Reader reader = new InputStreamReader(new BOMInputStream(contentStream)); final CSVParser csvParser = new CSVParser(reader, csvFormat)) { final List<RecordField> fields = new ArrayList<>(); for (final String columnName : csvParser.getHeaderMap().keySet()) { fields.add(new RecordField(columnName, RecordFieldType.STRING.getDataType(), true)); } return new SimpleRecordSchema(fields); } } catch (final Exception e) { throw new SchemaNotFoundException("Failed to read Header line from CSV", e); } }
CSVHeaderSchemaStrategy implements SchemaAccessStrategy { @Override public RecordSchema getSchema(Map<String, String> variables, final InputStream contentStream, final RecordSchema readSchema) throws SchemaNotFoundException { if (this.context == null) { throw new SchemaNotFoundException("Schema Access Strategy intended only for validation purposes and cannot obtain schema"); } try { final CSVFormat csvFormat = CSVUtils.createCSVFormat(context).withFirstRecordAsHeader(); try (final Reader reader = new InputStreamReader(new BOMInputStream(contentStream)); final CSVParser csvParser = new CSVParser(reader, csvFormat)) { final List<RecordField> fields = new ArrayList<>(); for (final String columnName : csvParser.getHeaderMap().keySet()) { fields.add(new RecordField(columnName, RecordFieldType.STRING.getDataType(), true)); } return new SimpleRecordSchema(fields); } } catch (final Exception e) { throw new SchemaNotFoundException("Failed to read Header line from CSV", e); } } }
CSVHeaderSchemaStrategy implements SchemaAccessStrategy { @Override public RecordSchema getSchema(Map<String, String> variables, final InputStream contentStream, final RecordSchema readSchema) throws SchemaNotFoundException { if (this.context == null) { throw new SchemaNotFoundException("Schema Access Strategy intended only for validation purposes and cannot obtain schema"); } try { final CSVFormat csvFormat = CSVUtils.createCSVFormat(context).withFirstRecordAsHeader(); try (final Reader reader = new InputStreamReader(new BOMInputStream(contentStream)); final CSVParser csvParser = new CSVParser(reader, csvFormat)) { final List<RecordField> fields = new ArrayList<>(); for (final String columnName : csvParser.getHeaderMap().keySet()) { fields.add(new RecordField(columnName, RecordFieldType.STRING.getDataType(), true)); } return new SimpleRecordSchema(fields); } } catch (final Exception e) { throw new SchemaNotFoundException("Failed to read Header line from CSV", e); } } CSVHeaderSchemaStrategy(final ConfigurationContext context);  CSVHeaderSchemaStrategy(final ValidationContext context); }
CSVHeaderSchemaStrategy implements SchemaAccessStrategy { @Override public RecordSchema getSchema(Map<String, String> variables, final InputStream contentStream, final RecordSchema readSchema) throws SchemaNotFoundException { if (this.context == null) { throw new SchemaNotFoundException("Schema Access Strategy intended only for validation purposes and cannot obtain schema"); } try { final CSVFormat csvFormat = CSVUtils.createCSVFormat(context).withFirstRecordAsHeader(); try (final Reader reader = new InputStreamReader(new BOMInputStream(contentStream)); final CSVParser csvParser = new CSVParser(reader, csvFormat)) { final List<RecordField> fields = new ArrayList<>(); for (final String columnName : csvParser.getHeaderMap().keySet()) { fields.add(new RecordField(columnName, RecordFieldType.STRING.getDataType(), true)); } return new SimpleRecordSchema(fields); } } catch (final Exception e) { throw new SchemaNotFoundException("Failed to read Header line from CSV", e); } } CSVHeaderSchemaStrategy(final ConfigurationContext context);  CSVHeaderSchemaStrategy(final ValidationContext context); @Override RecordSchema getSchema(Map<String, String> variables, final InputStream contentStream, final RecordSchema readSchema); @Override Set<SchemaField> getSuppliedSchemaFields(); }
CSVHeaderSchemaStrategy implements SchemaAccessStrategy { @Override public RecordSchema getSchema(Map<String, String> variables, final InputStream contentStream, final RecordSchema readSchema) throws SchemaNotFoundException { if (this.context == null) { throw new SchemaNotFoundException("Schema Access Strategy intended only for validation purposes and cannot obtain schema"); } try { final CSVFormat csvFormat = CSVUtils.createCSVFormat(context).withFirstRecordAsHeader(); try (final Reader reader = new InputStreamReader(new BOMInputStream(contentStream)); final CSVParser csvParser = new CSVParser(reader, csvFormat)) { final List<RecordField> fields = new ArrayList<>(); for (final String columnName : csvParser.getHeaderMap().keySet()) { fields.add(new RecordField(columnName, RecordFieldType.STRING.getDataType(), true)); } return new SimpleRecordSchema(fields); } } catch (final Exception e) { throw new SchemaNotFoundException("Failed to read Header line from CSV", e); } } CSVHeaderSchemaStrategy(final ConfigurationContext context);  CSVHeaderSchemaStrategy(final ValidationContext context); @Override RecordSchema getSchema(Map<String, String> variables, final InputStream contentStream, final RecordSchema readSchema); @Override Set<SchemaField> getSuppliedSchemaFields();  }