NIFI-15869 Add Unmatched Field Behavior property to PutBigQuery (#11172)

0BVer · web-flow · commit d8a2453eb568 · 2026-04-24T12:00:31.000+02:00
diff --git a/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/bigquery/PutBigQuery.java b/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/bigquery/PutBigQuery.java
@@ -172,6 +172,14 @@ public class PutBigQuery extends AbstractBigQueryProcessor {
         .defaultValue("false")
         .build();
 
+    public static final PropertyDescriptor UNMATCHED_FIELD_BEHAVIOR = new PropertyDescriptor.Builder()
+        .name("Unmatched Field Behavior")
+        .description("Specifies how to handle fields in the incoming record that do not exist in the destination BigQuery table schema.")
+        .required(true)
+        .allowableValues(UnmatchedFieldBehavior.class)
+        .defaultValue(UnmatchedFieldBehavior.IGNORE)
+        .build();
+
     private static final List<PropertyDescriptor> DESCRIPTORS = List.of(
         GCP_CREDENTIALS_PROVIDER_SERVICE,
         PROJECT_ID,
@@ -183,6 +191,7 @@ public class PutBigQuery extends AbstractBigQueryProcessor {
         APPEND_RECORD_COUNT,
         RETRY_COUNT,
         SKIP_INVALID_ROWS,
+        UNMATCHED_FIELD_BEHAVIOR,
         PROXY_CONFIGURATION_SERVICE
     );
 
@@ -235,13 +244,14 @@ public void onTrigger(ProcessContext context, ProcessSession session)  {
         }
 
         final boolean skipInvalidRows = context.getProperty(SKIP_INVALID_ROWS).evaluateAttributeExpressions(flowFile).asBoolean();
+        final UnmatchedFieldBehavior unmatchedFieldBehavior = context.getProperty(UNMATCHED_FIELD_BEHAVIOR).asAllowableValue(UnmatchedFieldBehavior.class);
         final RecordReaderFactory readerFactory = context.getProperty(RECORD_READER).asControllerService(RecordReaderFactory.class);
 
         int recordNumWritten;
         try {
             try (InputStream in = session.read(flowFile);
                     RecordReader reader = readerFactory.createRecordReader(flowFile, in, getLogger())) {
-                recordNumWritten = writeRecordsToStream(reader, protoDescriptor, skipInvalidRows, tableSchema);
+                recordNumWritten = writeRecordsToStream(reader, protoDescriptor, skipInvalidRows, tableSchema, unmatchedFieldBehavior, tableName);
             }
             flowFile = session.putAttribute(flowFile, JOB_NB_RECORDS_ATTR, Integer.toString(recordNumWritten));
         } catch (Exception e) {
@@ -261,13 +271,14 @@ public void migrateProperties(PropertyConfiguration config) {
         config.renameProperty("bq.skip.invalid.rows", SKIP_INVALID_ROWS.getName());
     }
 
-    private int writeRecordsToStream(RecordReader reader, Descriptors.Descriptor descriptor, boolean skipInvalidRows, TableSchema tableSchema) throws Exception {
+    private int writeRecordsToStream(final RecordReader reader, final Descriptors.Descriptor descriptor, final boolean skipInvalidRows, final TableSchema tableSchema,
+                                     final UnmatchedFieldBehavior unmatchedFieldBehavior, final TableName tableName) throws Exception {
         Record currentRecord;
         int offset = 0;
         int recordNum = 0;
         ProtoRows.Builder rowsBuilder = ProtoRows.newBuilder();
         while ((currentRecord = reader.nextRecord()) != null) {
-            DynamicMessage message = recordToProtoMessage(currentRecord, descriptor, skipInvalidRows, tableSchema);
+            final DynamicMessage message = recordToProtoMessage(currentRecord, descriptor, skipInvalidRows, tableSchema, unmatchedFieldBehavior, tableName);
 
             if (message == null) {
                 continue;
@@ -289,12 +300,30 @@ private int writeRecordsToStream(RecordReader reader, Descriptors.Descriptor des
         return recordNum;
     }
 
-    private DynamicMessage recordToProtoMessage(Record record, Descriptors.Descriptor descriptor, boolean skipInvalidRows, TableSchema tableSchema) {
-        Map<String, Object> valueMap = convertMapRecord(record.toMap(), tableSchema.getFieldsList());
+    private DynamicMessage recordToProtoMessage(final Record record, final Descriptors.Descriptor descriptor, final boolean skipInvalidRows, final TableSchema tableSchema,
+                                                final UnmatchedFieldBehavior unmatchedFieldBehavior, final TableName tableName) {
+        final Map<String, Object> rawMap = record.toMap();
+
+        if (unmatchedFieldBehavior != UnmatchedFieldBehavior.IGNORE) {
+            final List<String> unmatchedFields = new ArrayList<>();
+            collectUnmatchedFields(rawMap, tableSchema.getFieldsList(), null, unmatchedFields);
+
+            if (!unmatchedFields.isEmpty()) {
+                if (unmatchedFieldBehavior == UnmatchedFieldBehavior.WARN) {
+                    getLogger().warn("Record contains {} field(s) not present in BigQuery table schema for table {}: {}",
+                            unmatchedFields.size(), tableName, unmatchedFields);
+                } else {
+                    throw new ProcessException("Record contains fields not present in BigQuery table schema for table %s: %s"
+                            .formatted(tableName, unmatchedFields));
+                }
+            }
+        }
+
+        final Map<String, Object> valueMap = convertMapRecord(rawMap, tableSchema.getFieldsList());
         DynamicMessage message = null;
         try {
             message = ProtoUtils.createMessage(descriptor, valueMap, tableSchema);
-        } catch (RuntimeException e) {
+        } catch (final RuntimeException e) {
             getLogger().error("Cannot convert record to message", e);
             if (!skipInvalidRows) {
                 throw e;
@@ -304,6 +333,33 @@ private DynamicMessage recordToProtoMessage(Record record, Descriptors.Descripto
         return message;
     }
 
+    private static void collectUnmatchedFields(final Map<String, Object> recordMap, final List<TableFieldSchema> tableFields,
+                                               final String parentPath, final List<String> unmatchedFields) {
+        for (final Map.Entry<String, Object> entry : recordMap.entrySet()) {
+            final String key = entry.getKey();
+            final Object value = entry.getValue();
+            final String fieldPath = parentPath == null ? key : parentPath + "." + key;
+            final TableFieldSchema fieldSchema = findFieldSchema(tableFields, key);
+
+            if (fieldSchema == null) {
+                unmatchedFields.add(fieldPath);
+                continue;
+            }
+
+            if (fieldSchema.getType() == TableFieldSchema.Type.STRUCT) {
+                if (value instanceof MapRecord mapRecord) {
+                    collectUnmatchedFields(mapRecord.toMap(), fieldSchema.getFieldsList(), fieldPath, unmatchedFields);
+                } else if (value instanceof Object[] arrayValue) {
+                    for (final Object item : arrayValue) {
+                        if (item instanceof MapRecord mapRecordItem) {
+                            collectUnmatchedFields(mapRecordItem.toMap(), fieldSchema.getFieldsList(), fieldPath, unmatchedFields);
+                        }
+                    }
+                }
+            }
+        }
+    }
+
     private void append(AppendContext appendContext) throws Exception {
         if (error.get() != null) {
             throw error.get();
diff --git a/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/bigquery/UnmatchedFieldBehavior.java b/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/bigquery/UnmatchedFieldBehavior.java
@@ -0,0 +1,62 @@
+/*
+ * 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.processors.gcp.bigquery;
+
+import org.apache.nifi.components.DescribedValue;
+
+/**
+ * Strategy applied when an incoming record contains a field that does not exist in the
+ * target BigQuery table schema. The BigQuery Storage Write API does not surface this condition
+ * because NiFi drops unknown fields client-side before encoding to Protobuf; this enum allows
+ * operators to opt in to logging or failure semantics.
+ */
+public enum UnmatchedFieldBehavior implements DescribedValue {
+
+    IGNORE("Ignore Unmatched Fields", "Ignore Unmatched Fields",
+            "Any record field that does not map to a BigQuery table column is silently ignored."),
+    WARN("Warn on Unmatched Fields", "Warn on Unmatched Fields",
+            "Any record field that does not map to a BigQuery table column is ignored, but a warning is logged per affected record."),
+    FAIL("Fail on Unmatched Fields", "Fail on Unmatched Fields", """
+            If a record contains a field that does not map to a BigQuery table column, the FlowFile is routed to the failure relationship. \
+            This decision is independent of Skip Invalid Rows, which only governs row-level serialization failures.""");
+
+    private final String value;
+    private final String displayName;
+    private final String description;
+
+    UnmatchedFieldBehavior(final String value, final String displayName, final String description) {
+        this.value = value;
+        this.displayName = displayName;
+        this.description = description;
+    }
+
+    @Override
+    public String getValue() {
+        return value;
+    }
+
+    @Override
+    public String getDisplayName() {
+        return displayName;
+    }
+
+    @Override
+    public String getDescription() {
+        return description;
+    }
+}
diff --git a/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/bigquery/PutBigQueryTest.java b/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/bigquery/PutBigQueryTest.java
@@ -48,6 +48,8 @@
 import org.apache.nifi.proxy.ProxyConfiguration;
 import org.apache.nifi.reporting.InitializationException;
 import org.apache.nifi.schema.access.SchemaAccessUtils;
+import org.apache.nifi.util.LogMessage;
+import org.apache.nifi.util.MockComponentLog;
 import org.apache.nifi.util.PropertyMigrationResult;
 import org.apache.nifi.util.TestRunner;
 import org.apache.nifi.util.TestRunners;
@@ -80,6 +82,7 @@
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.ArgumentMatchers.isA;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.reset;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -367,6 +370,112 @@ void testUnknownColumnSkipped() {
         runner.assertAllFlowFilesTransferred(PutBigQuery.REL_SUCCESS);
     }
 
+    @Test
+    void testUnmatchedFieldWarnLogsAndWrites() {
+        when(writeClient.createWriteStream(isA(CreateWriteStreamRequest.class))).thenReturn(writeStream);
+        final TableSchema myTableSchema = mockTableSchema(FIELD_1_NAME, TableFieldSchema.Type.STRING, FIELD_2_NAME, TableFieldSchema.Type.STRING);
+        when(writeStream.getTableSchema()).thenReturn(myTableSchema);
+        when(streamWriter.append(isA(ProtoRows.class), isA(Long.class)))
+            .thenReturn(ApiFutures.immediateFuture(AppendRowsResponse.newBuilder().setAppendResult(mock(AppendRowsResponse.AppendResult.class)).build()));
+
+        runner.setProperty(PutBigQuery.UNMATCHED_FIELD_BEHAVIOR, UnmatchedFieldBehavior.WARN);
+
+        final String unknownProperty = "myUnknownProperty";
+        runner.enqueue(CSV_HEADER + ",unknownField\nmyId,myValue," + unknownProperty);
+        runner.run();
+
+        verify(streamWriter).append(protoRowsCaptor.capture(), offsetCaptor.capture());
+        final ProtoRows rows = protoRowsCaptor.getValue();
+        assertEquals(1, rows.getSerializedRowsCount());
+        assertFalse(rows.getSerializedRowsList().getFirst().toString().contains(unknownProperty));
+
+        runner.assertAllFlowFilesTransferred(PutBigQuery.REL_SUCCESS);
+        assertUnmatchedFieldWarningLogged(runner);
+    }
+
+    @Test
+    void testUnmatchedFieldFailRoutesToFailure() {
+        when(writeClient.createWriteStream(isA(CreateWriteStreamRequest.class))).thenReturn(writeStream);
+        final TableSchema myTableSchema = mockTableSchema(FIELD_1_NAME, TableFieldSchema.Type.STRING, FIELD_2_NAME, TableFieldSchema.Type.STRING);
+        when(writeStream.getTableSchema()).thenReturn(myTableSchema);
+
+        runner.setProperty(PutBigQuery.UNMATCHED_FIELD_BEHAVIOR, UnmatchedFieldBehavior.FAIL);
+        runner.setProperty(PutBigQuery.SKIP_INVALID_ROWS, "false");
+
+        runner.enqueue(CSV_HEADER + ",unknownField\nmyId,myValue,extraField");
+        runner.run();
+
+        verify(streamWriter, never()).append(any(ProtoRows.class), anyLong());
+        runner.assertAllFlowFilesTransferred(PutBigQuery.REL_FAILURE);
+    }
+
+    @Test
+    void testUnmatchedFieldFailIgnoresSkipInvalidRows() {
+        when(writeClient.createWriteStream(isA(CreateWriteStreamRequest.class))).thenReturn(writeStream);
+        final TableSchema myTableSchema = mockTableSchema(FIELD_1_NAME, TableFieldSchema.Type.STRING, FIELD_2_NAME, TableFieldSchema.Type.STRING);
+        when(writeStream.getTableSchema()).thenReturn(myTableSchema);
+
+        runner.setProperty(PutBigQuery.UNMATCHED_FIELD_BEHAVIOR, UnmatchedFieldBehavior.FAIL);
+        runner.setProperty(PutBigQuery.SKIP_INVALID_ROWS, "true");
+
+        runner.enqueue(CSV_HEADER + ",unknownField\nidOne,valueOne,extraField");
+        runner.run();
+
+        verify(streamWriter, never()).append(any(ProtoRows.class), anyLong());
+        runner.assertAllFlowFilesTransferred(PutBigQuery.REL_FAILURE);
+    }
+
+    @Test
+    void testUnmatchedFieldInNestedStructLogsDottedPath() throws InitializationException {
+        when(writeClient.createWriteStream(isA(CreateWriteStreamRequest.class))).thenReturn(writeStream);
+
+        final TableFieldSchema nestedKnown = mock(TableFieldSchema.class);
+        when(nestedKnown.getName()).thenReturn("known");
+        when(nestedKnown.getType()).thenReturn(TableFieldSchema.Type.STRING);
+        when(nestedKnown.getMode()).thenReturn(TableFieldSchema.Mode.NULLABLE);
+
+        final TableFieldSchema parent = mock(TableFieldSchema.class);
+        when(parent.getName()).thenReturn("parent");
+        when(parent.getType()).thenReturn(TableFieldSchema.Type.STRUCT);
+        when(parent.getMode()).thenReturn(TableFieldSchema.Mode.NULLABLE);
+        when(parent.getFieldsList()).thenReturn(List.of(nestedKnown));
+
+        final TableSchema schemaWithStruct = mock(TableSchema.class);
+        when(schemaWithStruct.getFieldsList()).thenReturn(List.of(parent));
+
+        when(writeStream.getTableSchema()).thenReturn(schemaWithStruct);
+        when(streamWriter.append(isA(ProtoRows.class), isA(Long.class)))
+            .thenReturn(ApiFutures.immediateFuture(AppendRowsResponse.newBuilder().setAppendResult(mock(AppendRowsResponse.AppendResult.class)).build()));
+
+        decorateWithNestedStructJsonReader(runner);
+
+        runner.setProperty(PutBigQuery.UNMATCHED_FIELD_BEHAVIOR, UnmatchedFieldBehavior.WARN);
+
+        runner.enqueue("""
+                {"parent":{"known":"ok","unexpected":"oops"}}
+                """);
+        runner.run();
+
+        runner.assertAllFlowFilesTransferred(PutBigQuery.REL_SUCCESS);
+        assertUnmatchedFieldPathLogged(runner, "parent.unexpected");
+    }
+
+    private static void assertUnmatchedFieldWarningLogged(final TestRunner runner) {
+        final MockComponentLog logger = runner.getLogger();
+        final boolean found = logger.getWarnMessages().stream()
+                .map(LogMessage::getMsg)
+                .anyMatch(msg -> msg != null && msg.contains("not present in BigQuery table schema"));
+        assertTrue(found, "Expected a warning about unmatched fields, got: " + logger.getWarnMessages());
+    }
+
+    private static void assertUnmatchedFieldPathLogged(final TestRunner runner, final String expectedPath) {
+        final MockComponentLog logger = runner.getLogger();
+        final boolean found = logger.getWarnMessages().stream()
+                .map(LogMessage::getMsg)
+                .anyMatch(msg -> msg != null && msg.contains(expectedPath));
+        assertTrue(found, "Expected warning to reference path '" + expectedPath + "', got: " + logger.getWarnMessages());
+    }
+
     @Test
     void testSchema() throws Exception {
         when(writeClient.createWriteStream(isA(CreateWriteStreamRequest.class))).thenReturn(writeStream);
@@ -561,6 +670,36 @@ private void decorateWithJsonRecordReaderWithSchema(TestRunner runner) throws In
         runner.enableControllerService(jsonReader);
     }
 
+    private void decorateWithNestedStructJsonReader(TestRunner runner) throws InitializationException {
+        String recordReaderSchema = """
+                {
+                  "name": "nested",
+                  "namespace": "nifi.examples",
+                  "type": "record",
+                  "fields": [
+                    {
+                      "name": "parent",
+                      "type": {
+                        "type": "record",
+                        "name": "ParentRecord",
+                        "fields": [
+                          {"name": "known", "type": ["null", "string"], "default": null},
+                          {"name": "unexpected", "type": ["null", "string"], "default": null}
+                        ]
+                      }
+                    }
+                  ]
+                }""";
+
+        JsonTreeReader jsonReader = new JsonTreeReader();
+        runner.addControllerService("nestedJsonReader", jsonReader);
+        runner.setProperty(jsonReader, SchemaAccessUtils.SCHEMA_ACCESS_STRATEGY, SchemaAccessUtils.SCHEMA_TEXT_PROPERTY);
+        runner.setProperty(jsonReader, SchemaAccessUtils.SCHEMA_TEXT, recordReaderSchema);
+        runner.enableControllerService(jsonReader);
+
+        runner.setProperty(PutBigQuery.RECORD_READER, "nestedJsonReader");
+    }
+
     private TableSchema mockTableSchema(String name1, TableFieldSchema.Type type1, String name2, TableFieldSchema.Type type2) {
         TableSchema myTableSchema = mock(TableSchema.class);
 
