NIFI-15587 - Add extensible FieldValidator and RecordValidator support to the Record API

Signed-off-by: Pierre Villard <pierre.villard.fr@gmail.com>

Author: pvillard31

nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/validation/DefaultValidationError.java

@@ -0,0 +1,131 @@
+/*
+ * 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.serialization.record.validation;
+
+import java.util.Objects;
+import java.util.Optional;
+import java.util.StringJoiner;
+
+/**
+ * Basic implementation of {@link ValidationError} that can be used by validators in modules that
+ * cannot depend on higher level utility classes. Instances are immutable and thread-safe.
+ */
+public class DefaultValidationError implements ValidationError {
+    private final Optional<String> fieldName;
+    private final Optional<Object> inputValue;
+    private final String explanation;
+    private final ValidationErrorType type;
+
+    private DefaultValidationError(final Builder builder) {
+        this.fieldName = Optional.ofNullable(builder.fieldName);
+        this.inputValue = Optional.ofNullable(builder.inputValue);
+        this.explanation = Objects.requireNonNull(builder.explanation, "Explanation is required");
+        this.type = Objects.requireNonNull(builder.type, "Validation error type is required");
+    }
+
+    @Override
+    public Optional<String> getFieldName() {
+        return fieldName;
+    }
+
+    @Override
+    public Optional<Object> getInputValue() {
+        return inputValue;
+    }
+
+    @Override
+    public String getExplanation() {
+        return explanation;
+    }
+
+    @Override
+    public ValidationErrorType getType() {
+        return type;
+    }
+
+    @Override
+    public String toString() {
+        final StringJoiner joiner = new StringJoiner(", ", "DefaultValidationError[", "]");
+        fieldName.ifPresent(name -> joiner.add("field=" + name));
+        inputValue.ifPresent(value -> joiner.add("value=" + value));
+        joiner.add("type=" + type);
+        joiner.add("explanation=" + explanation);
+        return joiner.toString();
+    }
+
+    @Override
+    public int hashCode() {
+        return 31 + 17 * fieldName.hashCode() + 17 * inputValue.hashCode() + 17 * explanation.hashCode();
+    }
+
+    @Override
+    public boolean equals(final Object obj) {
+        if (obj == this) {
+            return true;
+        }
+        if (obj == null) {
+            return false;
+        }
+        if (!(obj instanceof ValidationError other)) {
+            return false;
+        }
+        return getFieldName().equals(other.getFieldName()) && getInputValue().equals(other.getInputValue()) && getExplanation().equals(other.getExplanation());
+    }
+
+    /**
+     * Creates a builder for constructing immutable {@link DefaultValidationError} instances.
+     *
+     * @return builder instance
+     */
+    public static Builder builder() {
+        return new Builder();
+    }
+
+    public static final class Builder {
+        private String fieldName;
+        private Object inputValue;
+        private String explanation;
+        private ValidationErrorType type = ValidationErrorType.INVALID_FIELD;
+
+        private Builder() {
+        }
+
+        public Builder fieldName(final String fieldName) {
+            this.fieldName = fieldName;
+            return this;
+        }
+
+        public Builder inputValue(final Object inputValue) {
+            this.inputValue = inputValue;
+            return this;
+        }
+
+        public Builder explanation(final String explanation) {
+            this.explanation = explanation;
+            return this;
+        }
+
+        public Builder type(final ValidationErrorType type) {
+            this.type = type;
+            return this;
+        }
+
+        public DefaultValidationError build() {
+            return new DefaultValidationError(this);
+        }
+    }
+}

nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/validation/FieldValidator.java

@@ -0,0 +1,40 @@
+/*
+ * 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.serialization.record.validation;
+
+import java.util.Collection;
+
+/**
+ * Provides validation for an individual record field value. Field Validators are expected to be immutable and thread-safe.
+ * If a validator needs field metadata (such as the data type), it should capture that information at construction time.
+ */
+public interface FieldValidator {
+
+    /**
+     * Validates the provided value for a field at the given path.
+     *
+     * @param fieldPath the path of the field being validated (used for clear diagnostics)
+     * @param value the value of the field for the record currently being validated
+     * @return a collection of validation errors. The collection must be empty when the value is valid.
+     */
+    Collection<ValidationError> validate(String fieldPath, Object value);
+
+    /**
+     * @return a short human readable description of what the validator enforces
+     */
+    String getDescription();
+}

nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/validation/RecordValidator.java

@@ -0,0 +1,43 @@
+/*
+ * 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.serialization.record.validation;
+
+import org.apache.nifi.serialization.record.Record;
+
+import java.util.Collection;
+
+/**
+ * Provides validation logic for an entire {@link Record} instance.
+ * Record Validators are expected to be immutable and thread-safe.
+ * If a validator needs schema metadata, it should capture that information at construction time.
+ */
+public interface RecordValidator {
+
+    /**
+     * Validates the provided record.
+     *
+     * @param record the record instance to validate
+     * @param fieldPath the path within the overall document that identifies the record (root records use the empty string)
+     * @return a collection of validation errors. The collection must be empty when the record is valid.
+     */
+    Collection<ValidationError> validate(Record record, String fieldPath);
+
+    /**
+     * @return a short human readable description of what the validator enforces
+     */
+    String getDescription();
+}

nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/validation/SchemaValidators.java

@@ -0,0 +1,75 @@
+/*
+ * 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.serialization.record.validation;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Immutable container that carries field-level and record-level validators independently of the schema model.
+ * Field validators are keyed by field name. Record validators apply to the entire record.
+ * Nested validators mirror the schema tree: a RECORD-type or ARRAY-of-RECORD field maps to a child
+ * {@code SchemaValidators} that applies to the nested record.
+ */
+public class SchemaValidators {
+    public static final SchemaValidators EMPTY = new SchemaValidators(Map.of(), List.of(), Map.of());
+
+    private final Map<String, List<FieldValidator>> fieldValidators;
+    private final List<RecordValidator> recordValidators;
+    private final Map<String, SchemaValidators> nestedValidators;
+
+    public SchemaValidators(final Map<String, List<FieldValidator>> fieldValidators, final List<RecordValidator> recordValidators) {
+        this(fieldValidators, recordValidators, Map.of());
+    }
+
+    public SchemaValidators(final Map<String, List<FieldValidator>> fieldValidators, final List<RecordValidator> recordValidators,
+            final Map<String, SchemaValidators> nestedValidators) {
+        if (fieldValidators == null || fieldValidators.isEmpty()) {
+            this.fieldValidators = Map.of();
+        } else {
+            final Map<String, List<FieldValidator>> defensiveCopy = new HashMap<>(fieldValidators.size());
+            for (final Map.Entry<String, List<FieldValidator>> entry : fieldValidators.entrySet()) {
+                defensiveCopy.put(entry.getKey(), List.copyOf(entry.getValue()));
+            }
+            this.fieldValidators = Collections.unmodifiableMap(defensiveCopy);
+        }
+        this.recordValidators = recordValidators == null || recordValidators.isEmpty() ? List.of() : List.copyOf(recordValidators);
+        this.nestedValidators = nestedValidators == null || nestedValidators.isEmpty() ? Map.of() : Map.copyOf(nestedValidators);
+    }
+
+    public List<FieldValidator> getFieldValidators(final String fieldName) {
+        return fieldValidators.getOrDefault(fieldName, List.of());
+    }
+
+    public Map<String, List<FieldValidator>> getAllFieldValidators() {
+        return fieldValidators;
+    }
+
+    public List<RecordValidator> getRecordValidators() {
+        return recordValidators;
+    }
+
+    public SchemaValidators getNestedValidators(final String fieldName) {
+        return nestedValidators.getOrDefault(fieldName, EMPTY);
+    }
+
+    public boolean isEmpty() {
+        return fieldValidators.isEmpty() && recordValidators.isEmpty() && nestedValidators.isEmpty();
+    }
+}