You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by bb...@apache.org on 2026/10/20 18:23:43 UTC
(nifi) branch main updated: NIFI-16478 Preserved nested arrays when choosing a CHOICE sub-type (#11824)
This is an automated email from the ASF dual-hosted git repository.
bbende 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 17044171992 NIFI-16478 Preserved nested arrays when choosing a CHOICE sub-type (#11824)
17044171992 is described below
commit 1704417199295801b2c7aa6c016e3dfa9bbe76a2
Author: Mark Payne <ma...@hotmail.com>
AuthorDate: Tue Oct 20 14:23:36 2026 -0400
NIFI-16478 Preserved nested arrays when choosing a CHOICE sub-type (#11824)
When several CHOICE sub-types are compatible with a value and none is a clear match, DataTypeUtils.chooseDataType now prefers a sub-type that does not convert nested arrays, collections, or maps into Strings, rather than always selecting the first compatible sub-type.
Co-authored-by: Cursor <cu...@cursor.com>
---
.../serialization/record/util/DataTypeUtils.java | 106 +++++++++++++++++++--
.../serialization/record/TestDataTypeUtils.java | 27 ++++++
.../org/apache/nifi/json/TestWriteJsonResult.java | 52 ++++++++++
3 files changed, 175 insertions(+), 10 deletions(-)
diff --git a/nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/util/DataTypeUtils.java b/nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/util/DataTypeUtils.java
index edad8cf0e73..6407df7a8bb 100644
--- a/nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/util/DataTypeUtils.java
+++ b/nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/util/DataTypeUtils.java
@@ -348,19 +348,105 @@ public class DataTypeUtils {
}
}
- int nrOfCompatibleSimpleSubTypes = compatibleSimpleSubTypes.size();
+ if (compatibleSimpleSubTypes.isEmpty()) {
+ return null;
+ }
- final DataType chosenSimpleType;
- if (nrOfCompatibleSimpleSubTypes == 0) {
- chosenSimpleType = null;
- } else if (nrOfCompatibleSimpleSubTypes == 1) {
- chosenSimpleType = compatibleSimpleSubTypes.getFirst();
- } else {
- chosenSimpleType = findMostSuitableType(value, compatibleSimpleSubTypes, Function.identity())
- .orElse(compatibleSimpleSubTypes.getFirst());
+ if (compatibleSimpleSubTypes.size() == 1) {
+ return compatibleSimpleSubTypes.getFirst();
+ }
+
+ final Optional<DataType> mostSuitableType = findMostSuitableType(value, compatibleSimpleSubTypes, Function.identity());
+ if (mostSuitableType.isPresent()) {
+ return mostSuitableType.get();
+ }
+
+ // Prefer a type that does not flatten nested arrays, collections, or maps into strings
+ for (final DataType compatibleSubType : compatibleSimpleSubTypes) {
+ if (isCompatibleWithoutLosingStructure(value, compatibleSubType)) {
+ return compatibleSubType;
+ }
}
- return chosenSimpleType;
+ return compatibleSimpleSubTypes.getFirst();
+ }
+
+ /**
+ * Determines whether the given value can be converted to the given type without turning an array, collection, or map
+ * anywhere within the value into a String. The value is expected to already be compatible with the given type.
+ */
+ private static boolean isCompatibleWithoutLosingStructure(final Object value, final DataType dataType) {
+ if (value == null) {
+ return true;
+ }
+
+ switch (dataType.getFieldType()) {
+ case STRING -> {
+ if (value instanceof Byte[]) {
+ return true;
+ }
+
+ if (value instanceof Object[] || value instanceof Collection || value instanceof Map) {
+ return false;
+ }
+
+ return true;
+ }
+ case ARRAY -> {
+ final DataType elementType = ((ArrayDataType) dataType).getElementType();
+ if (elementType == null || !(value instanceof final Object[] array)) {
+ return true;
+ }
+
+ for (final Object element : array) {
+ if (!isCompatibleWithoutLosingStructure(element, elementType)) {
+ return false;
+ }
+ }
+
+ return true;
+ }
+ case RECORD -> {
+ final RecordSchema childSchema = ((RecordDataType) dataType).getChildSchema();
+ if (childSchema == null) {
+ return true;
+ }
+
+ for (final RecordField childField : childSchema.getFields()) {
+ final Object childValue;
+ if (value instanceof final Record record) {
+ childValue = record.getValue(childField);
+ } else if (value instanceof final Map<?, ?> map) {
+ childValue = map.get(childField.getFieldName());
+ } else {
+ return true;
+ }
+
+ if (!isCompatibleWithoutLosingStructure(childValue, childField.getDataType())) {
+ return false;
+ }
+ }
+
+ return true;
+ }
+ case MAP -> {
+ final DataType valueType = ((MapDataType) dataType).getValueType();
+ if (valueType == null || !(value instanceof final Map<?, ?> map)) {
+ return true;
+ }
+
+ for (final Object mapValue : map.values()) {
+ if (!isCompatibleWithoutLosingStructure(mapValue, valueType)) {
+ return false;
+ }
+ }
+
+ return true;
+ }
+ default -> {
+ return true;
+ }
+ }
}
public static <T> Optional<T> findMostSuitableType(Object value, List<T> types, Function<T, DataType> dataTypeMapper) {
diff --git a/nifi-commons/nifi-record/src/test/java/org/apache/nifi/serialization/record/TestDataTypeUtils.java b/nifi-commons/nifi-record/src/test/java/org/apache/nifi/serialization/record/TestDataTypeUtils.java
index 1ac591e8ef5..ac61c98949e 100644
--- a/nifi-commons/nifi-record/src/test/java/org/apache/nifi/serialization/record/TestDataTypeUtils.java
+++ b/nifi-commons/nifi-record/src/test/java/org/apache/nifi/serialization/record/TestDataTypeUtils.java
@@ -1298,6 +1298,33 @@ public class TestDataTypeUtils {
assertEquals(RecordFieldType.RECORD, ((ArrayDataType) chosenRecord).getElementType().getFieldType());
}
+ @Test
+ void testChooseDataTypeForNestedArraysInChoice() {
+ final RecordSchema entrySchema = new SimpleRecordSchema(List.of(
+ new RecordField("key", RecordFieldType.STRING.getDataType()),
+ new RecordField("value", RecordFieldType.STRING.getDataType())));
+ final MapRecord entry = new MapRecord(entrySchema, Map.of("key", "a", "value", "b"));
+
+ final DataType stringArray = RecordFieldType.ARRAY.getArrayDataType(RecordFieldType.STRING.getDataType());
+ final DataType recordArray = RecordFieldType.ARRAY.getArrayDataType(RecordFieldType.RECORD.getRecordDataType(entrySchema));
+ final DataType twoLevelRecordArray = RecordFieldType.ARRAY.getArrayDataType(recordArray);
+ final DataType threeLevelRecordArray = RecordFieldType.ARRAY.getArrayDataType(twoLevelRecordArray);
+ final List<DataType> choices = Arrays.asList(stringArray, threeLevelRecordArray);
+
+ final Object[] emptyBranch = new Object[] {new Object[0]};
+ final Object[] entries = new Object[] {entry};
+ final Object[] populatedBranch = new Object[] {entries};
+
+ final Object[] emptyAndPopulatedBranches = new Object[] {emptyBranch, populatedBranch};
+ testChooseDataTypeAlsoReverseTypes(emptyAndPopulatedBranches, choices, threeLevelRecordArray);
+
+ final Object[] onlyEmptyBranches = new Object[] {emptyBranch};
+ testChooseDataTypeAlsoReverseTypes(onlyEmptyBranches, choices, threeLevelRecordArray);
+
+ final Object[] strings = new Object[] {"a", "b"};
+ testChooseDataTypeAlsoReverseTypes(strings, choices, stringArray);
+ }
+
@Test
public void testMergeDataTypesMergesRecordSchemasInsteadOfCreatingChoice() {
final RecordSchema schemaA = new SimpleRecordSchema(List.of(
diff --git a/nifi-extension-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/json/TestWriteJsonResult.java b/nifi-extension-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/json/TestWriteJsonResult.java
index 572a9cd76c2..b20881ba5c0 100644
--- a/nifi-extension-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/json/TestWriteJsonResult.java
+++ b/nifi-extension-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/json/TestWriteJsonResult.java
@@ -17,6 +17,7 @@
package org.apache.nifi.json;
+import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.NullSuppression;
import org.apache.nifi.logging.ComponentLog;
@@ -31,6 +32,7 @@ import org.apache.nifi.serialization.record.RecordFieldType;
import org.apache.nifi.serialization.record.RecordSchema;
import org.apache.nifi.serialization.record.RecordSet;
import org.apache.nifi.serialization.record.SerializedForm;
+import org.apache.nifi.serialization.record.util.DataTypeUtils;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
@@ -56,6 +58,7 @@ import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
class TestWriteJsonResult {
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
@Test
public void testScientificNotationUsage() throws IOException {
@@ -712,6 +715,55 @@ class TestWriteJsonResult {
assertEquals(json, output);
}
+ @Test
+ void testChoiceArrayOfStringsOrNestedArraysOfRecordsFromMergedSchemas() throws IOException {
+ final String emptyJson = """
+ {"attributes":[]}""";
+ final String nestedJson = """
+ {
+ "attributes": [
+ [
+ []
+ ],
+ [
+ [
+ {
+ "key": "a",
+ "value": "b"
+ }
+ ]
+ ]
+ ]
+ }
+ """;
+
+ // Schemas inferred separately for an empty array and for nested arrays of records merge into a choice that must keep the nesting
+ final JsonSchemaInference jsonSchemaInference = new JsonSchemaInference(new TimeValueInference(null, null, null));
+ final RecordSchema emptySchema = jsonSchemaInference.inferSchema(new JsonRecordSource(new ByteArrayInputStream(emptyJson.getBytes(StandardCharsets.UTF_8))));
+ final RecordSchema nestedSchema = jsonSchemaInference.inferSchema(new JsonRecordSource(new ByteArrayInputStream(nestedJson.getBytes(StandardCharsets.UTF_8))));
+ final RecordSchema mergedSchema = DataTypeUtils.merge(emptySchema, nestedSchema);
+
+ final RecordSchema entrySchema = new SimpleRecordSchema(List.of(
+ new RecordField("key", RecordFieldType.STRING.getDataType()),
+ new RecordField("value", RecordFieldType.STRING.getDataType())));
+ final Record entry = new MapRecord(entrySchema, Map.of("key", "a", "value", "b"));
+ final Object[] emptyBranch = new Object[] {new Object[0]};
+ final Object[] entries = new Object[] {entry};
+ final Object[] populatedBranch = new Object[] {entries};
+ final Object[] attributes = new Object[] {emptyBranch, populatedBranch};
+ final Record nestedRecord = new MapRecord(nestedSchema, Map.of("attributes", attributes));
+
+ final ByteArrayOutputStream output = new ByteArrayOutputStream();
+ try (final WriteJsonResult writer = new WriteJsonResult(Mockito.mock(ComponentLog.class), mergedSchema, new SchemaNameAsAttribute(), output, false,
+ NullSuppression.NEVER_SUPPRESS, OutputGrouping.OUTPUT_ONELINE, null, null, null)) {
+ writer.beginRecordSet();
+ writer.writeRecord(nestedRecord);
+ writer.finishRecordSet();
+ }
+
+ assertEquals(OBJECT_MAPPER.readTree(nestedJson), OBJECT_MAPPER.readTree(output.toString(StandardCharsets.UTF_8)));
+ }
+
@Test
void testReuseInputSerializationDefaultTrueUsesFastPath() throws IOException {
final List<RecordField> fields = new ArrayList<>();