NIFI-15047 - Initial JSON Schemas support for NiFi Records

nifi-commons/nifi-json-schema-shared/pom.xml

@@ -31,5 +31,19 @@
             <artifactId>nifi-json-schema-api</artifactId>
             <version>2.7.0-SNAPSHOT</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.nifi</groupId>
+            <artifactId>nifi-record</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.nifi</groupId>
+            <artifactId>nifi-standard-record-utils</artifactId>
+            <version>${project.version}</version>
+        </dependency>
     </dependencies>
-</project>
+</project>

nifi-commons/nifi-json-schema-shared/src/main/java/org/apache/nifi/json/schema/validation/AdditionalPropertiesValidator.java

@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.json.schema.validation;
+
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.serialization.record.RecordSchema;
+import org.apache.nifi.serialization.record.validation.RecordValidator;
+import org.apache.nifi.serialization.record.validation.ValidationError;
+import org.apache.nifi.serialization.record.validation.ValidationErrorType;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Set;
+import java.util.regex.Pattern;
+
+/**
+ * Enforces the JSON Schema {@code additionalProperties: false} constraint.
+ */
+public class AdditionalPropertiesValidator implements RecordValidator {
+    private final Set<String> allowedFields;
+    private final List<Pattern> allowedPatterns;
+
+    public AdditionalPropertiesValidator(final Set<String> allowedFields) {
+        this(allowedFields, List.of());
+    }
+
+    public AdditionalPropertiesValidator(final Set<String> allowedFields, final Collection<Pattern> allowedPatterns) {
+        this.allowedFields = Set.copyOf(allowedFields);
+        this.allowedPatterns = List.copyOf(allowedPatterns);
+    }
+
+    @Override
+    public Collection<ValidationError> validate(final Record record, final RecordSchema schema, final String fieldPath) {
+        final Collection<ValidationError> errors = new ArrayList<>();
+        for (final String rawFieldName : record.getRawFieldNames()) {
+            if (!allowedFields.contains(rawFieldName) && allowedPatterns.stream().noneMatch(pattern -> pattern.matcher(rawFieldName).find())) {
+                final String fullFieldPath = ValidatorUtils.buildFieldPath(fieldPath, rawFieldName);
+                final ValidationError error = ValidatorUtils.createError(fullFieldPath, record.getValue(rawFieldName), ValidationErrorType.EXTRA_FIELD,
+                        "Field is not defined in schema");
+                errors.add(error);
+            }
+        }
+        return errors;
+    }
+
+    @Override
+    public String getDescription() {
+        return "Additional properties disallowed";
+    }
+}

nifi-commons/nifi-json-schema-shared/src/main/java/org/apache/nifi/json/schema/validation/ArrayLengthValidator.java

@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.json.schema.validation;
+
+import org.apache.nifi.serialization.record.RecordField;
+import org.apache.nifi.serialization.record.validation.FieldValidator;
+import org.apache.nifi.serialization.record.validation.ValidationError;
+import org.apache.nifi.serialization.record.validation.ValidationErrorType;
+
+import java.util.Collection;
+
+/**
+ * Validates {@code minItems} and {@code maxItems} constraints for array fields.
+ */
+public class ArrayLengthValidator implements FieldValidator {
+    private final Integer minItems;
+    private final Integer maxItems;
+
+    public ArrayLengthValidator(final Integer minItems, final Integer maxItems) {
+        this.minItems = minItems;
+        this.maxItems = maxItems;
+    }
+
+    @Override
+    public Collection<ValidationError> validate(final RecordField field, final String fieldPath, final Object value) {
+        final int size = determineSize(value);
+        if (size < 0) {
+            return ValidatorUtils.errorCollection(null);
+        }
+
+        if (minItems != null && size < minItems) {
+            final String explanation = String.format("Array must contain at least %d items", minItems);
+            final ValidationError error = ValidatorUtils.createError(fieldPath, value, ValidationErrorType.INVALID_FIELD, explanation);
+            return ValidatorUtils.errorCollection(error);
+        }
+
+        if (maxItems != null && size > maxItems) {
+            final String explanation = String.format("Array must contain no more than %d items", maxItems);
+            final ValidationError error = ValidatorUtils.createError(fieldPath, value, ValidationErrorType.INVALID_FIELD, explanation);
+            return ValidatorUtils.errorCollection(error);
+        }
+
+        return ValidatorUtils.errorCollection(null);
+    }
+
+    @Override
+    public String getDescription() {
+        final StringBuilder description = new StringBuilder("Array length validator");
+        if (minItems != null) {
+            description.append(", minItems=").append(minItems);
+        }
+        if (maxItems != null) {
+            description.append(", maxItems=").append(maxItems);
+        }
+        return description.toString();
+    }
+
+    private int determineSize(final Object value) {
+        if (value == null) {
+            return -1;
+        }
+
+        if (value instanceof Object[]) {
+            return ((Object[]) value).length;
+        }
+
+        if (value instanceof java.util.Collection<?>) {
+            return ((java.util.Collection<?>) value).size();
+        }
+
+        return -1;
+    }
+}

nifi-commons/nifi-json-schema-shared/src/main/java/org/apache/nifi/json/schema/validation/ConstValidator.java

@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.json.schema.validation;
+
+import org.apache.nifi.serialization.record.RecordField;
+import org.apache.nifi.serialization.record.validation.FieldValidator;
+import org.apache.nifi.serialization.record.validation.ValidationError;
+import org.apache.nifi.serialization.record.validation.ValidationErrorType;
+
+import java.util.Collection;
+
+/**
+ * Validates that the field value matches the JSON Schema {@code const} keyword.
+ */
+public class ConstValidator implements FieldValidator {
+    private final String canonicalValue;
+
+    public ConstValidator(final Object constantValue) {
+        this.canonicalValue = EnumValidator.canonicalize(constantValue);
+    }
+
+    @Override
+    public Collection<ValidationError> validate(final RecordField field, final String fieldPath, final Object value) {
+        final String candidate = EnumValidator.canonicalize(value);
+        if (!canonicalValue.equals(candidate)) {
+            final String explanation = String.format("Value must equal %s", canonicalValue);
+            final ValidationError error = ValidatorUtils.createError(fieldPath, value, ValidationErrorType.INVALID_FIELD, explanation);
+            return ValidatorUtils.errorCollection(error);
+        }
+
+        return ValidatorUtils.errorCollection(null);
+    }
+
+    @Override
+    public String getDescription() {
+        return "Const validator: value=" + canonicalValue;
+    }
+}

nifi-commons/nifi-json-schema-shared/src/main/java/org/apache/nifi/json/schema/validation/EnumValidator.java

@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.json.schema.validation;
+
+import org.apache.nifi.serialization.record.RecordField;
+import org.apache.nifi.serialization.record.validation.FieldValidator;
+import org.apache.nifi.serialization.record.validation.ValidationError;
+import org.apache.nifi.serialization.record.validation.ValidationErrorType;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Validates a value against a finite set of allowed values as defined by the JSON Schema {@code enum} keyword.
+ */
+public class EnumValidator implements FieldValidator {
+    private final Set<String> allowedValues;
+    private final String description;
+
+    public EnumValidator(final Collection<String> allowedValues) {
+        this.allowedValues = Set.copyOf(allowedValues);
+        this.description = "Enum validator" + allowedValues;
+    }
+
+    @Override
+    public Collection<ValidationError> validate(final RecordField field, final String fieldPath, final Object value) {
+        final String canonicalValue = canonicalize(value);
+        if (!allowedValues.contains(canonicalValue)) {
+            final String explanation = String.format("Value must be one of %s", allowedValues);
+            final ValidationError error = ValidatorUtils.createError(fieldPath, value, ValidationErrorType.INVALID_FIELD, explanation);
+            return ValidatorUtils.errorCollection(error);
+        }
+
+        return ValidatorUtils.errorCollection(null);
+    }
+
+    @Override
+    public String getDescription() {
+        return description;
+    }
+
+    public static Collection<String> canonicalizeAll(final Collection<Object> values) {
+        return values.stream().map(EnumValidator::canonicalizeStatic).collect(Collectors.toSet());
+    }
+
+    public static String canonicalize(final Object value) {
+        return canonicalizeStatic(value);
+    }
+
+    private static String canonicalizeStatic(final Object value) {
+        if (value == null) {
+            return "null";
+        }
+
+        final Class<?> valueClass = value.getClass();
+        if (valueClass.isArray()) {
+            if (value instanceof Object[]) {
+                return Arrays.deepToString((Object[]) value);
+            }
+            if (value instanceof byte[]) {
+                return Arrays.toString((byte[]) value);
+            }
+            if (value instanceof short[]) {
+                return Arrays.toString((short[]) value);
+            }
+            if (value instanceof int[]) {
+                return Arrays.toString((int[]) value);
+            }
+            if (value instanceof long[]) {
+                return Arrays.toString((long[]) value);
+            }
+            if (value instanceof float[]) {
+                return Arrays.toString((float[]) value);
+            }
+            if (value instanceof double[]) {
+                return Arrays.toString((double[]) value);
+            }
+            if (value instanceof char[]) {
+                return Arrays.toString((char[]) value);
+            }
+            if (value instanceof boolean[]) {
+                return Arrays.toString((boolean[]) value);
+            }
+        }
+
+        return String.valueOf(value);
+    }
+}