This is an automated email from the ASF dual-hosted git repository.
markap14 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 69f83555e95 NIFI-16467 Fixed exponential schema inference time for
deeply nested records (#11817)
69f83555e95 is described below
commit 69f83555e95712b65a12a80659b4a0f0a3b60e99
Author: kubaugust <[email protected]>
AuthorDate: Tue Oct 20 16:57:24 2026 +0200
NIFI-16467 Fixed exponential schema inference time for deeply nested
records (#11817)
* NIFI-16467 Fixed exponential schema inference time for deeply nested
records
Check nested RECORD and ARRAY field types in one direction in
DataTypeUtils.isRecordWider instead of calling getWiderType, which compares
nested records in both directions and doubles the cost at each nesting level
when neither record is wider.
* NIFI-16467 Added nested array nullability tests for wider record checks
* NIFI-16467 Addressed review feedback
---
.../serialization/record/util/DataTypeUtils.java | 32 ++++++++++++++++----
.../serialization/record/TestDataTypeUtils.java | 24 +++++++++++++++
.../apache/nifi/json/TestJsonSchemaInference.java | 34 +++++++++++++++++-----
3 files changed, 78 insertions(+), 12 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 d40fec78b44..edad8cf0e73 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
@@ -1922,17 +1922,39 @@ public class DataTypeUtils {
continue;
}
- final Optional<DataType> widerType =
getWiderType(widerField.get().getDataType(), narrowerField.getDataType());
- if (widerType.isEmpty()) {
+ if (!isWiderOrEqual(widerField.get().getDataType(),
narrowerField.getDataType())) {
return false;
}
+ }
- if (!widerType.get().equals(widerField.get().getDataType())) {
- return false;
+ return true;
+ }
+
+ // Nested records and arrays are not passed to getWiderType, which checks
both directions and doubles the cost at each nesting level
+ private static boolean isWiderOrEqual(final DataType potentiallyWider,
final DataType potentiallyNarrower) {
+ final RecordFieldType widerFieldType = potentiallyWider.getFieldType();
+ final RecordFieldType narrowerFieldType =
potentiallyNarrower.getFieldType();
+
+ if (widerFieldType == RecordFieldType.RECORD && narrowerFieldType ==
RecordFieldType.RECORD) {
+ final RecordSchema widerSchema = ((RecordDataType)
potentiallyWider).getChildSchema();
+ final RecordSchema narrowerSchema = ((RecordDataType)
potentiallyNarrower).getChildSchema();
+ if (widerSchema != null && narrowerSchema != null) {
+ return isRecordWider(widerSchema, narrowerSchema);
+ }
+ } else if (widerFieldType == RecordFieldType.ARRAY &&
narrowerFieldType == RecordFieldType.ARRAY) {
+ final DataType widerElementType = ((ArrayDataType)
potentiallyWider).getElementType();
+ final DataType narrowerElementType = ((ArrayDataType)
potentiallyNarrower).getElementType();
+ if (widerElementType != null && narrowerElementType != null) {
+ return isWiderOrEqual(widerElementType, narrowerElementType)
&& !((ArrayDataType) potentiallyWider).isElementsNullable();
}
}
- return true;
+ final Optional<DataType> widerType = getWiderType(potentiallyWider,
potentiallyNarrower);
+ if (widerType.isEmpty()) {
+ return false;
+ }
+
+ return widerType.get().equals(potentiallyWider);
}
private static boolean isDecimalType(final RecordFieldType fieldType) {
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 ac358c2c800..1ac591e8ef5 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
@@ -34,6 +34,7 @@ import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.sql.Date;
import java.sql.Types;
+import java.time.Duration;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
@@ -61,6 +62,7 @@ import static
org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
import static org.junit.jupiter.api.Assertions.assertTrue;
public class TestDataTypeUtils {
@@ -1688,4 +1690,26 @@ public class TestDataTypeUtils {
assertTrue(merged.getField("tags").orElseThrow().isNullable());
assertFalse(((ArrayDataType)
merged.getField("tags").orElseThrow().getDataType()).isElementsNullable());
}
+
+ @Test
+ public void
testMergeDataTypesWithDeeplyNestedNonWiderRecordsCompletesQuickly() {
+ final DataType merged =
assertTimeoutPreemptively(Duration.ofSeconds(10),
+ () -> DataTypeUtils.mergeDataTypes(createNestedRecordType(40,
"a"), createNestedRecordType(40, "b")));
+
+ final RecordSchema mergedSchema = ((RecordDataType)
merged).getChildSchema();
+ assertEquals(List.of("content", "a", "b"),
mergedSchema.getFieldNames());
+ }
+
+ private DataType createNestedRecordType(final int depth, final String...
fieldNames) {
+ DataType dataType = RecordFieldType.STRING.getDataType();
+ for (int i = 0; i < depth; i++) {
+ final List<RecordField> fields = new ArrayList<>();
+ fields.add(new RecordField("content",
RecordFieldType.ARRAY.getArrayDataType(dataType)));
+ for (final String fieldName : fieldNames) {
+ fields.add(new RecordField(fieldName,
RecordFieldType.STRING.getDataType()));
+ }
+ dataType = RecordFieldType.RECORD.getRecordDataType(new
SimpleRecordSchema(fields));
+ }
+ return dataType;
+ }
}
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/TestJsonSchemaInference.java
b/nifi-extension-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/json/TestJsonSchemaInference.java
index a4f2b8e79e2..5dfaef6941b 100644
---
a/nifi-extension-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/json/TestJsonSchemaInference.java
+++
b/nifi-extension-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/json/TestJsonSchemaInference.java
@@ -30,15 +30,19 @@ import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import java.io.BufferedInputStream;
+import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
import java.util.Arrays;
import java.util.List;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
import static org.junit.jupiter.api.Assertions.assertTrue;
class TestJsonSchemaInference {
@@ -178,17 +182,33 @@ class TestJsonSchemaInference {
assertSame(RecordFieldType.STRING, arrayElementType.getFieldType());
}
+ @Test
+ void testDeeplyNestedNonWiderRecordsCompleteQuickly() {
+ final String json = "[{\"doc\":[" + createNestedJson("a", 40) + "," +
createNestedJson("b", 40) + "]}]";
+
+ final RecordSchema schema =
assertTimeoutPreemptively(Duration.ofSeconds(10),
+ () -> inferSchema(new
ByteArrayInputStream(json.getBytes(StandardCharsets.UTF_8))));
+
+ final DataType elementType = ((ArrayDataType)
schema.getDataType("doc").get()).getElementType();
+ assertEquals(List.of("content", "a", "b"), ((RecordDataType)
elementType).getChildSchema().getFieldNames());
+ }
+
+ private String createNestedJson(final String fieldName, final int depth) {
+ return "{\"content\":[".repeat(depth) + "\"item\"" + ("],\"" +
fieldName + "\":\"item\"}").repeat(depth);
+ }
+
private RecordSchema inferSchema(final File jsonFile) throws IOException {
try (final InputStream in = new FileInputStream(jsonFile);
final InputStream bufferedIn = new BufferedInputStream(in)) {
-
- final InferSchemaAccessStrategy<?> accessStrategy = new
InferSchemaAccessStrategy<>(
- (variables, content) -> new JsonRecordSource(content),
- new JsonSchemaInference(timestampInference),
Mockito.mock(ComponentLog.class));
-
- final RecordSchema schema = accessStrategy.getSchema(null,
bufferedIn, null);
- return schema;
+ return inferSchema(bufferedIn);
}
+ }
+
+ private RecordSchema inferSchema(final InputStream in) throws IOException {
+ final InferSchemaAccessStrategy<?> accessStrategy = new
InferSchemaAccessStrategy<>(
+ (variables, content) -> new JsonRecordSource(content),
+ new JsonSchemaInference(timestampInference),
Mockito.mock(ComponentLog.class));
+ return accessStrategy.getSchema(null, in, null);
}
}
