@Test public void testFixedLengthComprehensive() throws Exception { int[] lengths = new int[] { 1, 9, 20, 12, 1, 2, 1, 3, 14, 14, 14, 14, 15, 9, 2, 3, 8, 4, 8, 7, 4, 11, 14, 14, 2, 1, 1, 1, 1, 1, 1, 1, 1, 6, 6, 6, 42, 1 }; List<Row> rows = new ArrayList<>(); StringBuilder sb = new StringBuilder(); for (int r = 0; r < 20; r++) { for (int i = 0; i < lengths.length; ++i) { sb.append(String.format("%1$" + lengths[i] + "s", "x")); } } rows.add(new Row("body", sb.toString())); String[] d = new String[] { "parse-as-fixed-length :body 1,9,20,12,1,2,1,3,14,14,14,14,15,9,2,3,8,4,8,7," + "4,11,14,14,2,1,1,1,1,1,1,1,1,6,6,6,42,1 ' '", "drop :body", "set columns A,B,C,D,E,F,G,H,I,J,K,L,M,N,O,P,Q,R,S,T,U,Z,W,X,Y,Z,A1,B1,C1,D1,E1,F1,G1,H1,I1,J1,K1,L1" }; RecipePipeline executor = TestingRig.execute(d); List<Row> results = executor.execute(rows); List<Row> errors = executor.errors(); Assert.assertEquals(20, results.size()); Assert.assertEquals(0, errors.size()); }
@Override public List<Row> execute(List<Row> rows, ExecutorContext context) throws DirectiveExecutionException, ErrorRowException { List<Row> results = new ArrayList<>(); for (Row row : rows) { int idx = row.find(col); if (idx != -1) { Object object = row.getValue(idx); if (object == null) { throw new DirectiveExecutionException( NAME, String.format("Column '%s' has null value. It should be a non-null 'String'.", col)); } if (object instanceof String) { String data = (String) object; int length = data.length(); if (length < recordLength) { throw new ErrorRowException( NAME, String.format("Column '%s' contains a value with fewer characters than the specified length " + "of row. Expected at least %d characters but found %s characters.", col, recordLength, length), 2); } int index = 1; while ((index + recordLength - 1) <= length) { Row newRow = new Row(row); int recPosition = index; int colid = 1; for (int width : widths) { String val = data.substring(recPosition - 1, recPosition + width - 1); if (padding != null) { val = val.replaceAll(padding, ""); } newRow.add(String.format("%s_%d", col, colid), val); recPosition += width; colid += 1; } results.add(newRow); index = (index + recordLength); } } else { throw new DirectiveExecutionException( NAME, String.format("Column '%s' has invalid type '%s'. It should be of type 'String'.", col, object.getClass().getSimpleName())); } } } return results; }
FixedLengthParser implements Directive, Lineage { @Override public List<Row> execute(List<Row> rows, ExecutorContext context) throws DirectiveExecutionException, ErrorRowException { List<Row> results = new ArrayList<>(); for (Row row : rows) { int idx = row.find(col); if (idx != -1) { Object object = row.getValue(idx); if (object == null) { throw new DirectiveExecutionException( NAME, String.format("Column '%s' has null value. It should be a non-null 'String'.", col)); } if (object instanceof String) { String data = (String) object; int length = data.length(); if (length < recordLength) { throw new ErrorRowException( NAME, String.format("Column '%s' contains a value with fewer characters than the specified length " + "of row. Expected at least %d characters but found %s characters.", col, recordLength, length), 2); } int index = 1; while ((index + recordLength - 1) <= length) { Row newRow = new Row(row); int recPosition = index; int colid = 1; for (int width : widths) { String val = data.substring(recPosition - 1, recPosition + width - 1); if (padding != null) { val = val.replaceAll(padding, ""); } newRow.add(String.format("%s_%d", col, colid), val); recPosition += width; colid += 1; } results.add(newRow); index = (index + recordLength); } } else { throw new DirectiveExecutionException( NAME, String.format("Column '%s' has invalid type '%s'. It should be of type 'String'.", col, object.getClass().getSimpleName())); } } } return results; } }
FixedLengthParser implements Directive, Lineage { @Override public List<Row> execute(List<Row> rows, ExecutorContext context) throws DirectiveExecutionException, ErrorRowException { List<Row> results = new ArrayList<>(); for (Row row : rows) { int idx = row.find(col); if (idx != -1) { Object object = row.getValue(idx); if (object == null) { throw new DirectiveExecutionException( NAME, String.format("Column '%s' has null value. It should be a non-null 'String'.", col)); } if (object instanceof String) { String data = (String) object; int length = data.length(); if (length < recordLength) { throw new ErrorRowException( NAME, String.format("Column '%s' contains a value with fewer characters than the specified length " + "of row. Expected at least %d characters but found %s characters.", col, recordLength, length), 2); } int index = 1; while ((index + recordLength - 1) <= length) { Row newRow = new Row(row); int recPosition = index; int colid = 1; for (int width : widths) { String val = data.substring(recPosition - 1, recPosition + width - 1); if (padding != null) { val = val.replaceAll(padding, ""); } newRow.add(String.format("%s_%d", col, colid), val); recPosition += width; colid += 1; } results.add(newRow); index = (index + recordLength); } } else { throw new DirectiveExecutionException( NAME, String.format("Column '%s' has invalid type '%s'. It should be of type 'String'.", col, object.getClass().getSimpleName())); } } } return results; }  }
FixedLengthParser implements Directive, Lineage { @Override public List<Row> execute(List<Row> rows, ExecutorContext context) throws DirectiveExecutionException, ErrorRowException { List<Row> results = new ArrayList<>(); for (Row row : rows) { int idx = row.find(col); if (idx != -1) { Object object = row.getValue(idx); if (object == null) { throw new DirectiveExecutionException( NAME, String.format("Column '%s' has null value. It should be a non-null 'String'.", col)); } if (object instanceof String) { String data = (String) object; int length = data.length(); if (length < recordLength) { throw new ErrorRowException( NAME, String.format("Column '%s' contains a value with fewer characters than the specified length " + "of row. Expected at least %d characters but found %s characters.", col, recordLength, length), 2); } int index = 1; while ((index + recordLength - 1) <= length) { Row newRow = new Row(row); int recPosition = index; int colid = 1; for (int width : widths) { String val = data.substring(recPosition - 1, recPosition + width - 1); if (padding != null) { val = val.replaceAll(padding, ""); } newRow.add(String.format("%s_%d", col, colid), val); recPosition += width; colid += 1; } results.add(newRow); index = (index + recordLength); } } else { throw new DirectiveExecutionException( NAME, String.format("Column '%s' has invalid type '%s'. It should be of type 'String'.", col, object.getClass().getSimpleName())); } } } return results; }  @Override UsageDefinition define(); @Override void initialize(Arguments args); @Override void destroy(); @Override List<Row> execute(List<Row> rows, ExecutorContext context); @Override Mutation lineage(); }
FixedLengthParser implements Directive, Lineage { @Override public List<Row> execute(List<Row> rows, ExecutorContext context) throws DirectiveExecutionException, ErrorRowException { List<Row> results = new ArrayList<>(); for (Row row : rows) { int idx = row.find(col); if (idx != -1) { Object object = row.getValue(idx); if (object == null) { throw new DirectiveExecutionException( NAME, String.format("Column '%s' has null value. It should be a non-null 'String'.", col)); } if (object instanceof String) { String data = (String) object; int length = data.length(); if (length < recordLength) { throw new ErrorRowException( NAME, String.format("Column '%s' contains a value with fewer characters than the specified length " + "of row. Expected at least %d characters but found %s characters.", col, recordLength, length), 2); } int index = 1; while ((index + recordLength - 1) <= length) { Row newRow = new Row(row); int recPosition = index; int colid = 1; for (int width : widths) { String val = data.substring(recPosition - 1, recPosition + width - 1); if (padding != null) { val = val.replaceAll(padding, ""); } newRow.add(String.format("%s_%d", col, colid), val); recPosition += width; colid += 1; } results.add(newRow); index = (index + recordLength); } } else { throw new DirectiveExecutionException( NAME, String.format("Column '%s' has invalid type '%s'. It should be of type 'String'.", col, object.getClass().getSimpleName())); } } } return results; }  @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; }