@Test public void testTrailingCommas() throws Exception { String[] directives = new String[] { "parse-as-csv body , false", "filter-rows-on regex-match body_1 ^school_id$", "drop body", "set columns school_id, student_id, last_name, first_name", "keep school_id,student_id,last_name,first_name" }; List<Row> rows = Arrays.asList( new Row("body", "school_id, student_id, last_name, first_name,,,"), new Row("body", "14J456,33445566,Potter,Harry,,,"), new Row("body", "14J456,44333433,Weasley,Ron,,,"), new Row("body", "14J456,65765566,Granger,Hermione,,,"), new Row("body", "14J456,13233121,Diggory,Cedric,,,"), new Row("body", "14J456,98786868,Weasley,George,,,"), new Row("body", "14J456,78977876,Weasley,Fred,,,") ); List<Row> expected = Arrays.asList( new Row("school_id", "14J456").add("student_id", "33445566").add("last_name", "Potter") .add("first_name", "Harry"), new Row("school_id", "14J456").add("student_id", "44333433").add("last_name", "Weasley") .add("first_name", "Ron"), new Row("school_id", "14J456").add("student_id", "65765566").add("last_name", "Granger") .add("first_name", "Hermione"), new Row("school_id", "14J456").add("student_id", "13233121").add("last_name", "Diggory") .add("first_name", "Cedric"), new Row("school_id", "14J456").add("student_id", "98786868").add("last_name", "Weasley") .add("first_name", "George"), new Row("school_id", "14J456").add("student_id", "78977876").add("last_name", "Weasley") .add("first_name", "Fred") ); rows = TestingRig.execute(directives, rows); Assert.assertEquals(6, rows.size()); for (int i = 0; i < rows.size(); ++i) { Assert.assertEquals(4, rows.get(i).width()); for (int j = 0; j < 4; j++) { Assert.assertEquals(expected.get(i).getValue(j), rows.get(i).getValue(j)); } } }
@Override public List<Row> execute(List<Row> rows, ExecutorContext context) throws DirectiveExecutionException, ErrorRowException { for (Row row : rows) { int idx = row.find(columnArg.value()); if (idx == -1) { continue; } String line = (String) row.getValue(idx); if (line == null || line.isEmpty()) { continue; } CSVParser parser = null; try { parser = CSVParser.parse(line, format); List<CSVRecord> csvRecords = parser.getRecords(); for (CSVRecord csvRecord : csvRecords) { if (!checkedHeader && hasHeader && isHeader(csvRecord)) { for (int i = 0; i < csvRecord.size(); i++) { headers.add(csvRecord.get(i).trim().replaceAll("\\s+", "_")); } if (rows.size() > 0) { return new ArrayList<>(); } } else { toRow(csvRecord, row); } } } catch (IOException e) { throw new ErrorRowException(NAME, e.getMessage(), 1); } } return rows; }
CsvParser implements Directive, Lineage { @Override public List<Row> execute(List<Row> rows, ExecutorContext context) throws DirectiveExecutionException, ErrorRowException { for (Row row : rows) { int idx = row.find(columnArg.value()); if (idx == -1) { continue; } String line = (String) row.getValue(idx); if (line == null || line.isEmpty()) { continue; } CSVParser parser = null; try { parser = CSVParser.parse(line, format); List<CSVRecord> csvRecords = parser.getRecords(); for (CSVRecord csvRecord : csvRecords) { if (!checkedHeader && hasHeader && isHeader(csvRecord)) { for (int i = 0; i < csvRecord.size(); i++) { headers.add(csvRecord.get(i).trim().replaceAll("\\s+", "_")); } if (rows.size() > 0) { return new ArrayList<>(); } } else { toRow(csvRecord, row); } } } catch (IOException e) { throw new ErrorRowException(NAME, e.getMessage(), 1); } } return rows; } }
CsvParser implements Directive, Lineage { @Override public List<Row> execute(List<Row> rows, ExecutorContext context) throws DirectiveExecutionException, ErrorRowException { for (Row row : rows) { int idx = row.find(columnArg.value()); if (idx == -1) { continue; } String line = (String) row.getValue(idx); if (line == null || line.isEmpty()) { continue; } CSVParser parser = null; try { parser = CSVParser.parse(line, format); List<CSVRecord> csvRecords = parser.getRecords(); for (CSVRecord csvRecord : csvRecords) { if (!checkedHeader && hasHeader && isHeader(csvRecord)) { for (int i = 0; i < csvRecord.size(); i++) { headers.add(csvRecord.get(i).trim().replaceAll("\\s+", "_")); } if (rows.size() > 0) { return new ArrayList<>(); } } else { toRow(csvRecord, row); } } } catch (IOException e) { throw new ErrorRowException(NAME, e.getMessage(), 1); } } return rows; }  }
CsvParser implements Directive, Lineage { @Override public List<Row> execute(List<Row> rows, ExecutorContext context) throws DirectiveExecutionException, ErrorRowException { for (Row row : rows) { int idx = row.find(columnArg.value()); if (idx == -1) { continue; } String line = (String) row.getValue(idx); if (line == null || line.isEmpty()) { continue; } CSVParser parser = null; try { parser = CSVParser.parse(line, format); List<CSVRecord> csvRecords = parser.getRecords(); for (CSVRecord csvRecord : csvRecords) { if (!checkedHeader && hasHeader && isHeader(csvRecord)) { for (int i = 0; i < csvRecord.size(); i++) { headers.add(csvRecord.get(i).trim().replaceAll("\\s+", "_")); } if (rows.size() > 0) { return new ArrayList<>(); } } else { toRow(csvRecord, row); } } } catch (IOException e) { throw new ErrorRowException(NAME, e.getMessage(), 1); } } return rows; }  @Override UsageDefinition define(); @Override void initialize(Arguments args); @Override void destroy(); @Override List<Row> execute(List<Row> rows, ExecutorContext context); @Override Mutation lineage(); }
CsvParser implements Directive, Lineage { @Override public List<Row> execute(List<Row> rows, ExecutorContext context) throws DirectiveExecutionException, ErrorRowException { for (Row row : rows) { int idx = row.find(columnArg.value()); if (idx == -1) { continue; } String line = (String) row.getValue(idx); if (line == null || line.isEmpty()) { continue; } CSVParser parser = null; try { parser = CSVParser.parse(line, format); List<CSVRecord> csvRecords = parser.getRecords(); for (CSVRecord csvRecord : csvRecords) { if (!checkedHeader && hasHeader && isHeader(csvRecord)) { for (int i = 0; i < csvRecord.size(); i++) { headers.add(csvRecord.get(i).trim().replaceAll("\\s+", "_")); } if (rows.size() > 0) { return new ArrayList<>(); } } else { toRow(csvRecord, row); } } } catch (IOException e) { throw new ErrorRowException(NAME, e.getMessage(), 1); } } return rows; }  @Override UsageDefinition define(); @Override void initialize(Arguments args); @Override void destroy(); @Override List<Row> execute(List<Row> rows, ExecutorContext context); @Override Mutation lineage(); static final String NAME; }