@Test public void testDepthParsing() throws Exception { String[] directives = new String[] { "parse-as-json body1 1", "parse-as-json body2 2", "parse-as-json body3 3", "parse-as-json body4 1", "parse-as-json body5 2", "parse-as-json body6 3", "parse-as-json body7 4" }; List<Row> rows = Arrays.asList( new Row("body1", "{\n" + " \"id\": 1,\n" + " \"name\": {\n" + " \"first\": \"Root\",\n" + " \"last\": \"Joltie\"\n" + " },\n" + " \"age\": 22,\n" + " \"weigth\": 184,\n" + " \"height\": 5.8\n" + "}").add("body2", "{\n" + " \"id\": 1,\n" + " \"name\": {\n" + " \"first\": \"Root\",\n" + " \"last\": \"Joltie\"\n" + " },\n" + " \"age\": 22,\n" + " \"weigth\": 184,\n" + " \"height\": 5.8\n" + "}").add("body3", "{\n" + " \"id\": 1,\n" + " \"name\": {\n" + " \"first\": \"Root\",\n" + " \"last\": \"Joltie\"\n" + " },\n" + " \"age\": 22,\n" + " \"weigth\": 184,\n" + " \"height\": 5.8\n" + "}").add("body4", "{\n" + " \"id\": 1,\n" + " \"name\": {\n" + " \"first\": {\n" + " \"n\" : \"Root\",\n" + " \"m\" : \"Rootie\"\n" + " },\n" + " \"last\": \"Joltie\"\n" + " },\n" + " \"age\": 22,\n" + " \"weigth\": 184,\n" + " \"height\": 5.8\n" + "}").add("body5", "{\n" + " \"id\": 1,\n" + " \"name\": {\n" + " \"first\": {\n" + " \"n\" : \"Root\",\n" + " \"m\" : \"Rootie\"\n" + " },\n" + " \"last\": \"Joltie\"\n" + " },\n" + " \"age\": 22,\n" + " \"weigth\": 184,\n" + " \"height\": 5.8\n" + "}").add("body6", "{\n" + " \"id\": 1,\n" + " \"name\": {\n" + " \"first\": {\n" + " \"n\" : \"Root\",\n" + " \"m\" : \"Rootie\"\n" + " },\n" + " \"last\": \"Joltie\"\n" + " },\n" + " \"age\": 22,\n" + " \"weigth\": 184,\n" + " \"height\": 5.8\n" + "}").add("body7", "{\n" + " \"id\": 1,\n" + " \"name\": {\n" + " \"first\": {\n" + " \"n\" : \"Root\",\n" + " \"m\" : \"Rootie\"\n" + " },\n" + " \"last\": \"Joltie\"\n" + " },\n" + " \"age\": 22,\n" + " \"weigth\": 184,\n" + " \"height\": 5.8\n" + "}") ); rows = TestingRig.execute(directives, rows); Assert.assertTrue(rows.size() == 1); Assert.assertNotEquals(rows.get(0).find("body1_name"), -1); Assert.assertNotEquals(rows.get(0).find("body2_name_first"), -1); Assert.assertNotEquals(rows.get(0).find("body2_name_last"), -1); Assert.assertNotEquals(rows.get(0).find("body3_name_first"), -1); Assert.assertNotEquals(rows.get(0).find("body6_name_first_n"), -1); Assert.assertNotEquals(rows.get(0).find("body6_name_first_m"), -1); }
@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(column); if (idx != -1) { Object value = row.getValue(idx); if (value == null) { continue; } try { JsonElement element = null; if (value instanceof String) { String document = (String) value; element = parser.parse(document.trim()); } else if (value instanceof JsonObject || value instanceof JsonArray) { element = (JsonElement) value; } else { throw new DirectiveExecutionException( NAME, String.format("Column '%s' is of invalid type '%s'. It should be of type 'String'" + " or 'JsonObject' or 'JsonArray'.", column, value.getClass().getSimpleName())); } row.remove(idx); if (element != null) { if (element instanceof JsonObject) { jsonFlatten(element.getAsJsonObject(), column, 1, depth, row); results.add(row); } else if (element instanceof JsonArray) { JsonArray array = element.getAsJsonArray(); if (array.size() > 0) { for (int i = 0; i < array.size(); ++i) { JsonElement object = array.get(i); Row newRow = new Row(row); newRow.add(column, getValue(object)); results.add(newRow); } } else { results.add(row); } } else if (element instanceof JsonPrimitive) { row.add(column, getValue(element.getAsJsonPrimitive())); } } } catch (JSONException e) { throw new ErrorRowException(NAME, e.getMessage(), 1); } } } return results; }
JsParser 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(column); if (idx != -1) { Object value = row.getValue(idx); if (value == null) { continue; } try { JsonElement element = null; if (value instanceof String) { String document = (String) value; element = parser.parse(document.trim()); } else if (value instanceof JsonObject || value instanceof JsonArray) { element = (JsonElement) value; } else { throw new DirectiveExecutionException( NAME, String.format("Column '%s' is of invalid type '%s'. It should be of type 'String'" + " or 'JsonObject' or 'JsonArray'.", column, value.getClass().getSimpleName())); } row.remove(idx); if (element != null) { if (element instanceof JsonObject) { jsonFlatten(element.getAsJsonObject(), column, 1, depth, row); results.add(row); } else if (element instanceof JsonArray) { JsonArray array = element.getAsJsonArray(); if (array.size() > 0) { for (int i = 0; i < array.size(); ++i) { JsonElement object = array.get(i); Row newRow = new Row(row); newRow.add(column, getValue(object)); results.add(newRow); } } else { results.add(row); } } else if (element instanceof JsonPrimitive) { row.add(column, getValue(element.getAsJsonPrimitive())); } } } catch (JSONException e) { throw new ErrorRowException(NAME, e.getMessage(), 1); } } } return results; } }
JsParser 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(column); if (idx != -1) { Object value = row.getValue(idx); if (value == null) { continue; } try { JsonElement element = null; if (value instanceof String) { String document = (String) value; element = parser.parse(document.trim()); } else if (value instanceof JsonObject || value instanceof JsonArray) { element = (JsonElement) value; } else { throw new DirectiveExecutionException( NAME, String.format("Column '%s' is of invalid type '%s'. It should be of type 'String'" + " or 'JsonObject' or 'JsonArray'.", column, value.getClass().getSimpleName())); } row.remove(idx); if (element != null) { if (element instanceof JsonObject) { jsonFlatten(element.getAsJsonObject(), column, 1, depth, row); results.add(row); } else if (element instanceof JsonArray) { JsonArray array = element.getAsJsonArray(); if (array.size() > 0) { for (int i = 0; i < array.size(); ++i) { JsonElement object = array.get(i); Row newRow = new Row(row); newRow.add(column, getValue(object)); results.add(newRow); } } else { results.add(row); } } else if (element instanceof JsonPrimitive) { row.add(column, getValue(element.getAsJsonPrimitive())); } } } catch (JSONException e) { throw new ErrorRowException(NAME, e.getMessage(), 1); } } } return results; }  }
JsParser 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(column); if (idx != -1) { Object value = row.getValue(idx); if (value == null) { continue; } try { JsonElement element = null; if (value instanceof String) { String document = (String) value; element = parser.parse(document.trim()); } else if (value instanceof JsonObject || value instanceof JsonArray) { element = (JsonElement) value; } else { throw new DirectiveExecutionException( NAME, String.format("Column '%s' is of invalid type '%s'. It should be of type 'String'" + " or 'JsonObject' or 'JsonArray'.", column, value.getClass().getSimpleName())); } row.remove(idx); if (element != null) { if (element instanceof JsonObject) { jsonFlatten(element.getAsJsonObject(), column, 1, depth, row); results.add(row); } else if (element instanceof JsonArray) { JsonArray array = element.getAsJsonArray(); if (array.size() > 0) { for (int i = 0; i < array.size(); ++i) { JsonElement object = array.get(i); Row newRow = new Row(row); newRow.add(column, getValue(object)); results.add(newRow); } } else { results.add(row); } } else if (element instanceof JsonPrimitive) { row.add(column, getValue(element.getAsJsonPrimitive())); } } } catch (JSONException e) { throw new ErrorRowException(NAME, e.getMessage(), 1); } } } 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 void jsonFlatten(JsonObject root, String field, int depth, int maxDepth, Row row); static Object getValue(JsonElement element); static Object getValue(JsonPrimitive primitive); }
JsParser 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(column); if (idx != -1) { Object value = row.getValue(idx); if (value == null) { continue; } try { JsonElement element = null; if (value instanceof String) { String document = (String) value; element = parser.parse(document.trim()); } else if (value instanceof JsonObject || value instanceof JsonArray) { element = (JsonElement) value; } else { throw new DirectiveExecutionException( NAME, String.format("Column '%s' is of invalid type '%s'. It should be of type 'String'" + " or 'JsonObject' or 'JsonArray'.", column, value.getClass().getSimpleName())); } row.remove(idx); if (element != null) { if (element instanceof JsonObject) { jsonFlatten(element.getAsJsonObject(), column, 1, depth, row); results.add(row); } else if (element instanceof JsonArray) { JsonArray array = element.getAsJsonArray(); if (array.size() > 0) { for (int i = 0; i < array.size(); ++i) { JsonElement object = array.get(i); Row newRow = new Row(row); newRow.add(column, getValue(object)); results.add(newRow); } } else { results.add(row); } } else if (element instanceof JsonPrimitive) { row.add(column, getValue(element.getAsJsonPrimitive())); } } } catch (JSONException e) { throw new ErrorRowException(NAME, e.getMessage(), 1); } } } 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 void jsonFlatten(JsonObject root, String field, int depth, int maxDepth, Row row); static Object getValue(JsonElement element); static Object getValue(JsonPrimitive primitive); static final String NAME; }