markap14 commented on a change in pull request #4934:
URL: https://github.com/apache/nifi/pull/4934#discussion_r604190380
##########
File path:
nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/json/TestJsonTreeRowRecordReader.java
##########
@@ -726,4 +732,383 @@ public void testIncorrectSchema() throws IOException,
MalformedRecordException {
assertTrue(msg.contains("Boolean"));
}
}
+
+ @Test
+ public void testMergeOfSimilarRecords() throws Exception {
+ // GIVEN
+ String jsonPath = "src/test/resources/json/similar-records.json";
+
+ RecordSchema expectedSchema = new SimpleRecordSchema(Arrays.asList(
+ new RecordField("integer", RecordFieldType.INT.getDataType()),
+ new RecordField("boolean", RecordFieldType.BOOLEAN.getDataType()),
+ new RecordField("booleanOrString",
RecordFieldType.CHOICE.getChoiceDataType(
+ RecordFieldType.BOOLEAN.getDataType(),
+ RecordFieldType.STRING.getDataType()
+ )),
+ new RecordField("string", RecordFieldType.STRING.getDataType())
+ ));
+
+ List<Object> expected = Arrays.asList(
+ new MapRecord(expectedSchema, new HashMap<String, Object>(){{
+ put("integer", 1);
+ put("boolean", true);
+ put("booleanOrString", true);
+ }}),
+ new MapRecord(expectedSchema, new HashMap<String, Object>(){{
+ put("integer", 2);
+ put("string", "stringValue2");
+ put("booleanOrString", "booleanOrStringValue2");
+ }})
+ );
+
+ // WHEN
+ // THEN
+ testReadRecords(jsonPath, expected);
+ }
+
+ @Test
+ public void testChoiceOfEmbeddedSimilarRecords() throws Exception {
+ // GIVEN
+ String jsonPath =
"src/test/resources/json/choice-of-embedded-similar-records.json";
+
+ SimpleRecordSchema expectedRecordSchema1 = new
SimpleRecordSchema(Arrays.asList(
+ new RecordField("integer", RecordFieldType.INT.getDataType()),
+ new RecordField("boolean", RecordFieldType.BOOLEAN.getDataType())
+ ));
+ SimpleRecordSchema expectedRecordSchema2 = new
SimpleRecordSchema(Arrays.asList(
+ new RecordField("integer", RecordFieldType.INT.getDataType()),
+ new RecordField("string", RecordFieldType.STRING.getDataType())
+ ));
+ RecordSchema expectedRecordChoiceSchema = new
SimpleRecordSchema(Arrays.asList(
+ new RecordField("record", RecordFieldType.CHOICE.getChoiceDataType(
+
RecordFieldType.RECORD.getRecordDataType(expectedRecordSchema1),
+ RecordFieldType.RECORD.getRecordDataType(expectedRecordSchema2)
+ ))
+ ));
+
+ List<Object> expected = Arrays.asList(
+ new MapRecord(expectedRecordChoiceSchema, new HashMap<String,
Object>(){{
+ put("record", new MapRecord(expectedRecordSchema1, new
HashMap<String, Object>(){{
+ put("integer", 1);
+ put("boolean", true);
+ }}));
+ }}),
+ new MapRecord(expectedRecordChoiceSchema, new HashMap<String,
Object>(){{
+ put("record", new MapRecord(expectedRecordSchema2, new
HashMap<String, Object>(){{
+ put("integer", 2);
+ put("string", "stringValue2");
+ }}));
+ }})
+ );
+
+ // WHEN
+ // THEN
+ testReadRecords(jsonPath, expected);
+ }
+
+ @Test
+ public void testChoiceOfEmbeddedArraysAndSingleRecords() throws Exception {
+ // GIVEN
+ String jsonPath =
"src/test/resources/json/choice-of-embedded-arrays-and-single-records.json";
+
+ SimpleRecordSchema expectedRecordSchema1 = new
SimpleRecordSchema(Arrays.asList(
+ new RecordField("integer", RecordFieldType.INT.getDataType())
+ ));
+ SimpleRecordSchema expectedRecordSchema2 = new
SimpleRecordSchema(Arrays.asList(
+ new RecordField("integer", RecordFieldType.INT.getDataType()),
+ new RecordField("boolean", RecordFieldType.BOOLEAN.getDataType())
+ ));
+ SimpleRecordSchema expectedRecordSchema3 = new
SimpleRecordSchema(Arrays.asList(
+ new RecordField("integer", RecordFieldType.INT.getDataType()),
+ new RecordField("string", RecordFieldType.STRING.getDataType())
+ ));
+ SimpleRecordSchema expectedRecordSchema4 = new
SimpleRecordSchema(Arrays.asList(
+ new RecordField("integer", RecordFieldType.INT.getDataType()),
+ new RecordField("string", RecordFieldType.STRING.getDataType())
+ ));
+ RecordSchema expectedRecordChoiceSchema = new
SimpleRecordSchema(Arrays.asList(
+ new RecordField("record", RecordFieldType.CHOICE.getChoiceDataType(
+
RecordFieldType.RECORD.getRecordDataType(expectedRecordSchema1),
+
RecordFieldType.RECORD.getRecordDataType(expectedRecordSchema3),
+
RecordFieldType.ARRAY.getArrayDataType(RecordFieldType.RECORD.getRecordDataType(expectedRecordSchema2)),
+
RecordFieldType.ARRAY.getArrayDataType(RecordFieldType.RECORD.getRecordDataType(expectedRecordSchema4))
+ ))
+ ));
+
+ List<Object> expected = Arrays.asList(
+ new MapRecord(expectedRecordChoiceSchema, new HashMap<String,
Object>(){{
+ put("record", new MapRecord(expectedRecordSchema1, new
HashMap<String, Object>(){{
+ put("integer", 1);
+ }}));
+ }}),
+ new MapRecord(expectedRecordChoiceSchema, new HashMap<String,
Object>(){{
+ put("record", new Object[]{
+ new MapRecord(expectedRecordSchema2, new HashMap<String,
Object>() {{
+ put("integer", 21);
+ put("boolean", true);
+ }}),
+ new MapRecord(expectedRecordSchema2, new HashMap<String,
Object>() {{
+ put("integer", 22);
+ put("boolean", false);
+ }})
+ });
+ }}),
+ new MapRecord(expectedRecordChoiceSchema, new HashMap<String,
Object>(){{
+ put("record", new MapRecord(expectedRecordSchema3, new
HashMap<String, Object>(){{
+ put("integer", 3);
+ put("string", "stringValue3");
+ }}));
+ }}),
+ new MapRecord(expectedRecordChoiceSchema, new HashMap<String,
Object>(){{
+ put("record", new Object[]{
+ new MapRecord(expectedRecordSchema4, new HashMap<String,
Object>() {{
+ put("integer", 41);
+ put("string", "stringValue41");
+ }}),
+ new MapRecord(expectedRecordSchema4, new HashMap<String,
Object>() {{
+ put("integer", 42);
+ put("string", "stringValue42");
+ }})
+ });
+ }})
+ );
+
+ // WHEN
+ // THEN
+ testReadRecords(jsonPath, expected);
+ }
+
+ @Test
+ public void testChoiceOfMergedEmbeddedArraysAndSingleRecords() throws
Exception {
+ // GIVEN
+ String jsonPath =
"src/test/resources/json/choice-of-merged-embedded-arrays-and-single-records.json";
+
+ SimpleRecordSchema expectedRecordSchema1 = new
SimpleRecordSchema(Arrays.asList(
+ new RecordField("integer", RecordFieldType.INT.getDataType()),
+ new RecordField("boolean", RecordFieldType.BOOLEAN.getDataType())
+ ));
+ SimpleRecordSchema expectedRecordSchema2 = new
SimpleRecordSchema(Arrays.asList(
+ new RecordField("integer", RecordFieldType.INT.getDataType()),
+ new RecordField("boolean", RecordFieldType.BOOLEAN.getDataType())
+ ));
+ SimpleRecordSchema expectedRecordSchema3 = new
SimpleRecordSchema(Arrays.asList(
+ new RecordField("integer", RecordFieldType.INT.getDataType()),
+ new RecordField("string", RecordFieldType.STRING.getDataType())
+ ));
+ SimpleRecordSchema expectedRecordSchema4 = new
SimpleRecordSchema(Arrays.asList(
+ new RecordField("integer", RecordFieldType.INT.getDataType()),
+ new RecordField("string", RecordFieldType.STRING.getDataType()),
+ new RecordField("boolean", RecordFieldType.BOOLEAN.getDataType())
+ ));
+ RecordSchema expectedRecordChoiceSchema = new
SimpleRecordSchema(Arrays.asList(
+ new RecordField("record", RecordFieldType.CHOICE.getChoiceDataType(
+
RecordFieldType.RECORD.getRecordDataType(expectedRecordSchema1),
+
RecordFieldType.RECORD.getRecordDataType(expectedRecordSchema3),
+
RecordFieldType.ARRAY.getArrayDataType(RecordFieldType.RECORD.getRecordDataType(expectedRecordSchema2)),
+
RecordFieldType.ARRAY.getArrayDataType(RecordFieldType.RECORD.getRecordDataType(expectedRecordSchema4))
+ ))
+ ));
+
+ List<Object> expected = Arrays.asList(
+ new MapRecord(expectedRecordChoiceSchema, new HashMap<String,
Object>(){{
+ put("record", new MapRecord(expectedRecordSchema1, new
HashMap<String, Object>(){{
+ put("integer", 1);
+ put("boolean", false);
+ }}));
+ }}),
+ new MapRecord(expectedRecordChoiceSchema, new HashMap<String,
Object>(){{
+ put("record", new Object[]{
+ new MapRecord(expectedRecordSchema2, new HashMap<String,
Object>() {{
+ put("integer", 21);
+ put("boolean", true);
+ }}),
+ new MapRecord(expectedRecordSchema2, new HashMap<String,
Object>() {{
+ put("integer", 22);
+ put("boolean", false);
+ }})
+ });
+ }}),
+ new MapRecord(expectedRecordChoiceSchema, new HashMap<String,
Object>(){{
+ put("record", new MapRecord(expectedRecordSchema3, new
HashMap<String, Object>(){{
+ put("integer", 3);
+ put("string", "stringValue3");
+ }}));
+ }}),
+ new MapRecord(expectedRecordChoiceSchema, new HashMap<String,
Object>(){{
+ put("record", new Object[]{
+ new MapRecord(expectedRecordSchema4, new HashMap<String,
Object>() {{
+ put("integer", 41);
+ put("string", "stringValue41");
+ }}),
+ new MapRecord(expectedRecordSchema4, new HashMap<String,
Object>() {{
+ put("integer", 42);
+ put("string", "stringValue42");
+ }}),
+ new MapRecord(expectedRecordSchema4, new HashMap<String,
Object>() {{
+ put("integer", 43);
+ put("boolean", false);
+ }})
+ });
+ }})
+ );
+
+ // WHEN
+ // THEN
+ testReadRecords(jsonPath, expected);
+ }
+
+ @Test
+ public void testChoseSuboptimalSchemaWhenDataHasExtraFields() throws
Exception {
+ // GIVEN
+ String jsonPath =
"src/test/resources/json/choice-of-different-arrays-with-extra-fields.json";
+
+ SimpleRecordSchema recordSchema1 = new
SimpleRecordSchema(Arrays.asList(
+ new RecordField("integer", RecordFieldType.INT.getDataType()),
+ new RecordField("boolean", RecordFieldType.BOOLEAN.getDataType())
+ ));
+ SimpleRecordSchema recordSchema2 = new
SimpleRecordSchema(Arrays.asList(
+ new RecordField("integer", RecordFieldType.INT.getDataType()),
+ new RecordField("string", RecordFieldType.STRING.getDataType())
+ ));
+
+ RecordSchema recordChoiceSchema = new SimpleRecordSchema(Arrays.asList(
+ new RecordField("record", RecordFieldType.CHOICE.getChoiceDataType(
+
RecordFieldType.ARRAY.getArrayDataType(RecordFieldType.RECORD.getRecordDataType(recordSchema1)),
+
RecordFieldType.ARRAY.getArrayDataType(RecordFieldType.RECORD.getRecordDataType(recordSchema2))
+ ))
+ ));
+
+ RecordSchema schema = new SimpleRecordSchema(Arrays.asList(
+ new RecordField("dataCollection",
RecordFieldType.ARRAY.getArrayDataType(
+ RecordFieldType.RECORD.getRecordDataType(recordChoiceSchema)
+ )
+ )));
+
+ SimpleRecordSchema expectedChildSchema1 = new
SimpleRecordSchema(Arrays.asList(
+ new RecordField("integer", RecordFieldType.INT.getDataType()),
+ new RecordField("boolean", RecordFieldType.BOOLEAN.getDataType())
+ ));
+ SimpleRecordSchema expectedChildSchema2 = new
SimpleRecordSchema(Arrays.asList(
+ new RecordField("integer", RecordFieldType.INT.getDataType()),
+ new RecordField("string", RecordFieldType.STRING.getDataType())
+ ));
+ RecordSchema expectedRecordChoiceSchema = new
SimpleRecordSchema(Arrays.asList(
+ new RecordField("record", RecordFieldType.CHOICE.getChoiceDataType(
+
RecordFieldType.ARRAY.getArrayDataType(RecordFieldType.RECORD.getRecordDataType(expectedChildSchema1)),
+
RecordFieldType.ARRAY.getArrayDataType(RecordFieldType.RECORD.getRecordDataType(expectedChildSchema2))
+ ))
+ ));
+
+ // Since the actual arrays have records with either (INT, BOOLEAN,
STRING) or (INT, STRING, STRING)
+ // while the explicit schema defines only (INT, BOOLEAN) and (INT,
STRING) we can't tell which record schema to chose
+ // so we take the first one (INT, BOOLEAN) - as best effort - for
both cases
+ SimpleRecordSchema expectedSelectedRecordSchemaForRecordsInBothArrays
= expectedChildSchema1;
+
+ List<Object> expected = Arrays.asList(
+ new MapRecord(expectedRecordChoiceSchema, new HashMap<String,
Object>(){{
+ put("record", new Object[]{
+ new
MapRecord(expectedSelectedRecordSchemaForRecordsInBothArrays, new
HashMap<String, Object>() {{
+ put("integer", 11);
+ put("boolean", true);
+ put("extraString", "extraStringValue11");
+ }}),
+ new
MapRecord(expectedSelectedRecordSchemaForRecordsInBothArrays, new
HashMap<String, Object>() {{
+ put("integer", 12);
+ put("boolean", false);
+ put("extraString", "extraStringValue12");
+ }})
+ });
+ }}),
+ new MapRecord(expectedRecordChoiceSchema, new HashMap<String,
Object>(){{
+ put("record", new Object[]{
+ new
MapRecord(expectedSelectedRecordSchemaForRecordsInBothArrays, new
HashMap<String, Object>() {{
+ put("integer", 21);
+ put("extraString", "extraStringValue21");
+ put("string", "stringValue21");
+ }}),
+ new
MapRecord(expectedSelectedRecordSchemaForRecordsInBothArrays, new
HashMap<String, Object>() {{
+ put("integer", 22);
+ put("extraString", "extraStringValue22");
+ put("string", "stringValue22");
+ }})
+ });
+ }})
+ );
+
+ // WHEN
+ // THEN
+ testReadRecords(jsonPath, schema, expected);
+ }
+
+ private void testReadRecords(String jsonPath, List<Object> expected)
throws IOException, MalformedRecordException {
+ // GIVEN
+ final File jsonFile = new File(jsonPath);
+
+ try (
+ InputStream jsonStream = new
ByteArrayInputStream(FileUtils.readFileToByteArray(jsonFile));
+ ) {
+ RecordSchema schema = inferSchema(jsonStream);
+
+ // WHEN
+ // THEN
+ testReadRecords(jsonStream, schema, expected);
+ }
+ }
+
+ private void testReadRecords(String jsonPath, RecordSchema schema,
List<Object> expected) throws IOException, MalformedRecordException {
+ // GIVEN
+ final File jsonFile = new File(jsonPath);
+
+ try (
+ InputStream jsonStream = new
ByteArrayInputStream(FileUtils.readFileToByteArray(jsonFile));
+ ) {
+ // WHEN
+ // THEN
+ testReadRecords(jsonStream, schema, expected);
+ }
+ }
+
+ private void testReadRecords(InputStream jsonStream, RecordSchema schema,
List<Object> expected) throws IOException, MalformedRecordException {
+ // GIVEN
+ try (
+ JsonTreeRowRecordReader reader = new
JsonTreeRowRecordReader(jsonStream, mock(ComponentLog.class), schema,
dateFormat, timeFormat, timestampFormat);
+ ) {
+ // WHEN
+ List<Object> actual = new ArrayList<>();
+ Record record;
+ while ((record = reader.nextRecord()) != null) {
+ List<Object> dataCollection = Arrays.asList((Object[])
record.getValue("dataCollection"));
+ actual.addAll(dataCollection);
+ }
+
+ // THEN
+ List<Function<Object, Object>> propertyProviders = Arrays.asList(
Review comment:
I find this logic very confusing, to have a List of functions that
transform one object, chained together, and then to wrap an existing list with
these transforms, just to compare them against the result of applying the same
chain of transformations.
It would be far simpler to just transform the objects here and then make the
assertion against the transformed objects.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
