@Test public void testExtraCommasAndLeadingZeros() 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, body_5", "set-column :last_name exp:{ this.width() == 5 ? (last_name + ',' + first_name) : last_name}", "set-column :first_name exp:{ this.width() == 5 ? body_5 : first_name}", "drop body_5" }; List<Row> rows = Arrays.asList( new Row("body", "school_id, student_id, last_name, first_name"), new Row("body", "14J456,0033445566,Potter,Jr,Harry"), new Row("body", "14J456,0044333433,Weasley,Ron") ); rows = TestingRig.execute(directives, rows); Assert.assertEquals(2, rows.size()); Assert.assertEquals(4, rows.get(0).width()); Assert.assertEquals(4, rows.get(1).width()); Assert.assertEquals("Potter,Jr", rows.get(0).getValue("last_name")); Assert.assertEquals("Harry", rows.get(0).getValue("first_name")); Assert.assertEquals("Weasley", rows.get(1).getValue("last_name")); Assert.assertEquals("Ron", rows.get(1).getValue("first_name")); Assert.assertEquals("0033445566", rows.get(0).getValue("student_id")); Assert.assertEquals("0044333433", rows.get(1).getValue("student_id")); }
@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; }