This is an automated email from the ASF dual-hosted git repository.
pvillard pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 37c2284 NIFI-8068: Ensure that when we determine the best of multiple
possible types in a UNION that we handle Arrays of Records properly. Also
refactored code to be a bit cleaner by extracting blocks of it into
appropriately named methods
37c2284 is described below
commit 37c2284d04ccf7ceb0d8ede62dafe6dce5ab835e
Author: Mark Payne <[email protected]>
AuthorDate: Thu Dec 3 14:34:21 2020 -0500
NIFI-8068: Ensure that when we determine the best of multiple possible
types in a UNION that we handle Arrays of Records properly. Also refactored
code to be a bit cleaner by extracting blocks of it into appropriately named
methods
Signed-off-by: Pierre Villard <[email protected]>
This closes #4706.
---
.../nifi-record-serialization-services/pom.xml | 2 +
.../nifi/json/AbstractJsonRowRecordReader.java | 111 ++++++++++++---------
.../nifi/json/TestJsonTreeRowRecordReader.java | 33 ++++++
.../json/choice-of-string-or-array-record.avsc | 29 ++++++
.../json/choice-of-string-or-array-record.json | 6 ++
5 files changed, 136 insertions(+), 45 deletions(-)
diff --git
a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/pom.xml
b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/pom.xml
index 3d01ea8..e35e957 100755
---
a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/pom.xml
+++
b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/pom.xml
@@ -170,6 +170,8 @@
<exclude>src/test/resources/json/record-choice.avsc</exclude>
<exclude>src/test/resources/json/prov-events.json</exclude>
<exclude>src/test/resources/json/docs-example.json</exclude>
+
<exclude>src/test/resources/json/choice-of-string-or-array-record.json</exclude>
+
<exclude>src/test/resources/json/choice-of-string-or-array-record.avsc</exclude>
<exclude>src/test/resources/syslog/syslog5424/log.txt</exclude>
<exclude>src/test/resources/syslog/syslog5424/log_all.txt</exclude>
<exclude>src/test/resources/syslog/syslog5424/log_mix.txt</exclude>
diff --git
a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/AbstractJsonRowRecordReader.java
b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/AbstractJsonRowRecordReader.java
index 6acb0d7..5f24431 100644
---
a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/AbstractJsonRowRecordReader.java
+++
b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/AbstractJsonRowRecordReader.java
@@ -191,72 +191,93 @@ public abstract class AbstractJsonRowRecordReader
implements RecordReader {
if (fieldNode.isObject()) {
RecordSchema childSchema = null;
if (dataType != null && RecordFieldType.MAP ==
dataType.getFieldType()) {
- final MapDataType mapDataType = (MapDataType) dataType;
- final DataType valueType = mapDataType.getValueType();
+ return getMapFromRawValue(fieldNode, dataType, fieldName);
+ }
- final Map<String, Object> mapValue = new HashMap<>();
+ return getRecordFromRawValue(fieldNode, dataType);
+ }
- final Iterator<Map.Entry<String, JsonNode>> fieldItr =
fieldNode.getFields();
- while (fieldItr.hasNext()) {
- final Map.Entry<String, JsonNode> entry = fieldItr.next();
- final String elementName = entry.getKey();
- final JsonNode elementNode = entry.getValue();
+ return null;
+ }
- final Object nodeValue = getRawNodeValue(elementNode,
valueType, fieldName + "['" + elementName + "']");
- mapValue.put(elementName, nodeValue);
- }
+ private Map<String, Object> getMapFromRawValue(final JsonNode fieldNode,
final DataType dataType, final String fieldName) throws IOException {
+ if (dataType == null || dataType.getFieldType() !=
RecordFieldType.MAP) {
+ return null;
+ }
- return mapValue;
- } else if (dataType != null && RecordFieldType.RECORD ==
dataType.getFieldType()) {
- final RecordDataType recordDataType = (RecordDataType)
dataType;
- childSchema = recordDataType.getChildSchema();
- } else if (dataType != null && RecordFieldType.CHOICE ==
dataType.getFieldType()) {
- final ChoiceDataType choiceDataType = (ChoiceDataType)
dataType;
+ final MapDataType mapDataType = (MapDataType) dataType;
+ final DataType valueType = mapDataType.getValueType();
- for (final DataType possibleDataType :
choiceDataType.getPossibleSubTypes()) {
- if (possibleDataType.getFieldType() !=
RecordFieldType.RECORD) {
- continue;
- }
+ final Map<String, Object> mapValue = new HashMap<>();
- final RecordSchema possibleSchema = ((RecordDataType)
possibleDataType).getChildSchema();
+ final Iterator<Map.Entry<String, JsonNode>> fieldItr =
fieldNode.getFields();
+ while (fieldItr.hasNext()) {
+ final Map.Entry<String, JsonNode> entry = fieldItr.next();
+ final String elementName = entry.getKey();
+ final JsonNode elementNode = entry.getValue();
- final Map<String, Object> childValues = new HashMap<>();
- final Iterator<String> fieldNames =
fieldNode.getFieldNames();
- while (fieldNames.hasNext()) {
- final String childFieldName = fieldNames.next();
+ final Object nodeValue = getRawNodeValue(elementNode, valueType,
fieldName + "['" + elementName + "']");
+ mapValue.put(elementName, nodeValue);
+ }
- final Object childValue =
getRawNodeValue(fieldNode.get(childFieldName),
possibleSchema.getDataType(childFieldName).orElse(null), childFieldName);
- childValues.put(childFieldName, childValue);
- }
+ return mapValue;
+ }
- final Record possibleRecord = new
MapRecord(possibleSchema, childValues);
- if (DataTypeUtils.isCompatibleDataType(possibleRecord,
possibleDataType)) {
- return possibleRecord;
- }
+ private Record getRecordFromRawValue(final JsonNode fieldNode, final
DataType dataType) throws IOException {
+ RecordSchema childSchema = null;
+ if (dataType != null && RecordFieldType.RECORD ==
dataType.getFieldType()) {
+ final RecordDataType recordDataType = (RecordDataType) dataType;
+ childSchema = recordDataType.getChildSchema();
+ } else if (dataType != null && RecordFieldType.CHOICE ==
dataType.getFieldType()) {
+ final ChoiceDataType choiceDataType = (ChoiceDataType) dataType;
+
+ for (final DataType possibleDataType :
choiceDataType.getPossibleSubTypes()) {
+ final Record record = createOptionalRecord(fieldNode,
possibleDataType);
+ if (record != null) {
+ return record;
}
}
+ }
- if (childSchema == null) {
- childSchema = new SimpleRecordSchema(Collections.emptyList());
- }
+ if (childSchema == null) {
+ childSchema = new SimpleRecordSchema(Collections.emptyList());
+ }
- final Iterator<String> fieldNames = fieldNode.getFieldNames();
- final Map<String, Object> childValues = new HashMap<>();
- while (fieldNames.hasNext()) {
- final String childFieldName = fieldNames.next();
+ return createRecordFromRawValue(fieldNode, childSchema);
+ }
- final DataType childDataType =
childSchema.getDataType(childFieldName).orElse(null);
- final Object childValue =
getRawNodeValue(fieldNode.get(childFieldName), childDataType, childFieldName);
- childValues.put(childFieldName, childValue);
- }
+ private Record createOptionalRecord(final JsonNode fieldNode, final
DataType dataType) throws IOException {
+ if (dataType.getFieldType() == RecordFieldType.RECORD) {
+ final RecordSchema possibleSchema = ((RecordDataType)
dataType).getChildSchema();
+ final Record possibleRecord = createRecordFromRawValue(fieldNode,
possibleSchema);
- final MapRecord record = new MapRecord(childSchema, childValues);
+ if (DataTypeUtils.isCompatibleDataType(possibleRecord, dataType)) {
+ return possibleRecord;
+ }
+ } else if (dataType.getFieldType() == RecordFieldType.ARRAY) {
+ final ArrayDataType arrayDataType = (ArrayDataType) dataType;
+ final DataType elementType = arrayDataType.getElementType();
+ final Record record = createOptionalRecord(fieldNode, elementType);
return record;
}
return null;
}
+ private Record createRecordFromRawValue(final JsonNode fieldNode, final
RecordSchema childSchema) throws IOException {
+ final Iterator<String> fieldNames = fieldNode.getFieldNames();
+ final Map<String, Object> childValues = new HashMap<>();
+ while (fieldNames.hasNext()) {
+ final String childFieldName = fieldNames.next();
+
+ final DataType childDataType =
childSchema.getDataType(childFieldName).orElse(null);
+ final Object childValue =
getRawNodeValue(fieldNode.get(childFieldName), childDataType, childFieldName);
+ childValues.put(childFieldName, childValue);
+ }
+
+ final MapRecord record = new MapRecord(childSchema, childValues);
+ return record;
+ }
protected JsonNode getNextJsonNode() throws IOException,
MalformedRecordException {
if (!firstObjectConsumed) {
diff --git
a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/json/TestJsonTreeRowRecordReader.java
b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/json/TestJsonTreeRowRecordReader.java
index bea213f..76562fa 100644
---
a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/json/TestJsonTreeRowRecordReader.java
+++
b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/json/TestJsonTreeRowRecordReader.java
@@ -28,6 +28,7 @@ import org.apache.nifi.serialization.record.RecordField;
import org.apache.nifi.serialization.record.RecordFieldType;
import org.apache.nifi.serialization.record.RecordSchema;
import org.apache.nifi.serialization.record.type.ChoiceDataType;
+import org.apache.nifi.util.MockComponentLog;
import org.junit.Assert;
import org.junit.Ignore;
import org.junit.Test;
@@ -87,6 +88,38 @@ public class TestJsonTreeRowRecordReader {
@Test
+ public void testReadChoiceOfStringOrArrayOfRecords() throws IOException,
MalformedRecordException {
+ final File schemaFile = new
File("src/test/resources/json/choice-of-string-or-array-record.avsc");
+ final File jsonFile = new
File("src/test/resources/json/choice-of-string-or-array-record.json");
+
+ final Schema avroSchema = new Schema.Parser().parse(schemaFile);
+ final RecordSchema recordSchema =
AvroTypeUtil.createSchema(avroSchema);
+
+ try (final InputStream fis = new FileInputStream(jsonFile);
+ final JsonTreeRowRecordReader reader = new
JsonTreeRowRecordReader(fis, new MockComponentLog("id", "id"), recordSchema,
dateFormat, timeFormat, timestampFormat)) {
+
+ final Record record = reader.nextRecord();
+ final Object[] fieldsArray = record.getAsArray("fields");
+ assertEquals(2, fieldsArray.length);
+
+ final Object firstElement = fieldsArray[0];
+ assertTrue(firstElement instanceof Record);
+ assertEquals("string", ((Record)
firstElement).getAsString("type"));
+
+ final Object secondElement = fieldsArray[1];
+ assertTrue(secondElement instanceof Record);
+ final Object[] typeArray = ((Record)
secondElement).getAsArray("type");
+ assertEquals(1, typeArray.length);
+
+ final Object firstType = typeArray[0];
+ assertTrue(firstType instanceof Record);
+ final Record firstTypeRecord = (Record) firstType;
+ assertEquals("string", firstTypeRecord.getAsString("type"));
+ }
+
+ }
+
+ @Test
@Ignore("Intended only for manual testing to determine performance
before/after modifications")
public void testPerformanceOnLocalFile() throws IOException,
MalformedRecordException {
final RecordSchema schema = new
SimpleRecordSchema(Collections.emptyList());
diff --git
a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/resources/json/choice-of-string-or-array-record.avsc
b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/resources/json/choice-of-string-or-array-record.avsc
new file mode 100644
index 0000000..bc38eb6
--- /dev/null
+++
b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/resources/json/choice-of-string-or-array-record.avsc
@@ -0,0 +1,29 @@
+{
+ "type" : "record",
+ "name" : "nifiRecord",
+ "namespace" : "org.apache.nifi",
+ "fields" : [ {
+ "name" : "fields",
+ "type" : [ "null", {
+ "type" : "array",
+ "items" : {
+ "type" : "record",
+ "name" : "fieldsType",
+ "fields" : [ {
+ "name" : "type",
+ "type" : [ "string", {
+ "type" : "array",
+ "items" : {
+ "type" : "record",
+ "name" : "typeType",
+ "fields" : [ {
+ "name" : "type",
+ "type" : [ "null", "string" ]
+ } ]
+ }
+ }, "null" ]
+ } ]
+ }
+ } ]
+ } ]
+}
\ No newline at end of file
diff --git
a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/resources/json/choice-of-string-or-array-record.json
b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/resources/json/choice-of-string-or-array-record.json
new file mode 100644
index 0000000..d7bee32
--- /dev/null
+++
b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/resources/json/choice-of-string-or-array-record.json
@@ -0,0 +1,6 @@
+{
+ "fields": [
+ {"type": "string"},
+ {"type": [{"type": "string"}]}
+ ]
+}
\ No newline at end of file
