MorphlineUtils { public static Row convertToRow(StructType schema, Record record) { LOG.debug("Converting Record to Row: {}", record); List<Object> values = Lists.newArrayList(); ListMultimap<String, Object> activeFields = record.getFields(); for (StructField field : schema.fields()) { String fieldName = field.name(); DataType fieldDataType = field.dataType(); Object recordValue = null; if (activeFields.containsKey(fieldName)) { recordValue = record.getFirstValue(fieldName); } if (LOG.isTraceEnabled()) { LOG.trace("Converting Field[{} => {}] to DataType[{}]]", fieldName, recordValue, fieldDataType); } if (null != recordValue) { try { Object result = RowUtils.toRowValue(recordValue, field.dataType()); values.add(result); } catch (Exception e) { throw new RuntimeException(String.format("Error converting Field[%s => %s] to DataType[%s]", fieldName, recordValue, fieldDataType), e); } } else { if (field.nullable()) { LOG.trace("Setting Field[{} => null] for DataType[{}]", fieldName, fieldDataType); values.add(null); } else { throw new RuntimeException(String.format("Error converting Field[%s => null] for DataType[%s]: DataType " + "cannot contain 'null'", fieldName, fieldDataType)); } } } Row result = new RowWithSchema(schema, values.toArray()); LOG.trace("Converted Record to Row: {}", result); return result; }  static Pipeline getPipeline(String morphlineFile, String morphlineId); static Pipeline setPipeline(String morphlineFile, String morphlineId, Collector collector, boolean isProduction); static List<Record> executePipeline(Pipeline pipeline, Record inputRecord); static List<Record> executePipeline(Pipeline pipeline, Record inputRecord, boolean errorOnEmpty); @SuppressWarnings("serial") static FlatMapFunction<Row, Row> morphlineMapper(final String morphlineFile, final String morphlineId,
                                                          final StructType outputSchema, final boolean errorOnEmpty); static Row convertToRow(StructType schema, Record record);  }
@Test public void convertToRowValidValue( final @Mocked RowUtils utils ) throws Exception { Record record = new Record(); record.put("field1", "one"); StructType schema = DataTypes.createStructType(Lists.newArrayList( DataTypes.createStructField("field1", DataTypes.StringType, false)) ); new Expectations() {{ RowUtils.toRowValue("one", DataTypes.StringType); result = "success"; }}; assertEquals("Invalid conversion", "success", MorphlineUtils.convertToRow(schema, record).get(0)); }
@Test public void convertToRowValidNullValue( final @Mocked RowUtils utils ) throws Exception { Record record = new Record(); record.put("field1", null); StructType schema = DataTypes.createStructType(Lists.newArrayList( DataTypes.createStructField("field1", DataTypes.StringType, true)) ); assertEquals("Invalid conversion", null, MorphlineUtils.convertToRow(schema, record).get(0)); new Verifications() {{ RowUtils.toRowValue(any, (DataType) any); times = 0; }}; }
@Test public void convertToRowInvalidNullValue( final @Mocked RowUtils utils ) throws Exception { Record record = new Record(); record.put("field1", null); StructType schema = DataTypes.createStructType(Lists.newArrayList( DataTypes.createStructField("field1", DataTypes.StringType, false)) ); try { MorphlineUtils.convertToRow(schema, record); fail("Did not throw a RuntimeException"); } catch (Exception e) { assertThat(e.getMessage(), CoreMatchers.containsString("DataType cannot contain 'null'")); } new Verifications() {{ RowUtils.toRowValue(any, (DataType) any); times = 0; }}; }
@Test public void convertToRowInvalidTypeNotNullable( final @Mocked RowUtils utils ) throws Exception { Record record = new Record(); record.put("field1", "one"); StructType schema = DataTypes.createStructType(Lists.newArrayList( DataTypes.createStructField("field1", DataTypes.StringType, false)) ); new Expectations() {{ RowUtils.toRowValue("one", DataTypes.StringType); result = new RuntimeException("Conversion exception"); }}; try { MorphlineUtils.convertToRow(schema, record); fail("Did not throw a RuntimeException"); } catch (Exception e) { assertThat(e.getMessage(), CoreMatchers.containsString("Error converting Field")); } }
@Test public void convertToRowInvalidTypeNullable( final @Mocked RowUtils utils ) throws Exception { Record record = new Record(); record.put("field1", "one"); StructType schema = DataTypes.createStructType(Lists.newArrayList( DataTypes.createStructField("field1", DataTypes.StringType, true)) ); new Expectations() {{ RowUtils.toRowValue("one", DataTypes.StringType); result = new RuntimeException("Conversion exception"); }}; try { MorphlineUtils.convertToRow(schema, record); fail("Did not throw a RuntimeException"); } catch (Exception e) { assertThat(e.getMessage(), CoreMatchers.containsString("Error converting Field")); } }
@Test public void convertToRowMissingColumnNotNullable( final @Mocked RowUtils utils ) throws Exception { Record record = new Record(); record.put("foo", "one"); StructType schema = DataTypes.createStructType(Lists.newArrayList( DataTypes.createStructField("field1", DataTypes.StringType, false)) ); try { MorphlineUtils.convertToRow(schema, record); fail("Did not throw a RuntimeException"); } catch (Exception e) { assertThat(e.getMessage(), CoreMatchers.containsString("DataType cannot contain 'null'")); } new Verifications() {{ RowUtils.toRowValue(any, (DataType) any); times = 0; }}; }
@Test public void convertToRowMissingColumnNullable( final @Mocked RowUtils utils ) throws Exception { Record record = new Record(); record.put("foo", "one"); StructType schema = DataTypes.createStructType(Lists.newArrayList( DataTypes.createStructField("field1", DataTypes.StringType, true)) ); MorphlineUtils.convertToRow(schema, record); new Verifications() {{ RowUtils.toRowValue(any, (DataType) any); times = 0; }}; }