mark-bathori commented on code in PR #7421:
URL: https://github.com/apache/nifi/pull/7421#discussion_r1342676660
##########
nifi-nar-bundles/nifi-iceberg-bundle/nifi-iceberg-common/src/main/java/org/apache/nifi/processors/iceberg/converter/IcebergRecordConverter.java:
##########
@@ -156,8 +174,25 @@ public DataType fieldPartner(DataType dataType, int
fieldId, String name) {
final RecordTypeWithFieldNameMapper recordType =
(RecordTypeWithFieldNameMapper) dataType;
final Optional<String> mappedFieldName =
recordType.getNameMapping(name);
- Validate.isTrue(mappedFieldName.isPresent(), String.format("Cannot
find field with name '%s' in the record schema", name));
-
+ if
(UnmatchedColumnBehavior.FAIL_UNMATCHED_COLUMN.equals(unmatchedColumnBehavior))
{
+ Validate.isTrue(mappedFieldName.isPresent(),
String.format("Cannot find field with name '%s' in the record schema", name));
+ }
+ if (mappedFieldName.isEmpty()) {
+ if
(UnmatchedColumnBehavior.WARNING_UNMATCHED_COLUMN.equals(unmatchedColumnBehavior))
{
+ if (logger != null) {
+ logger.warn("Cannot find field with name '" + name +
"' in the record schema, using the target schema for datatype and a null
value");
+ }
+ }
+ // If the field is missing, use the expected type from the
schema (converted to a DataType)
+ final Types.NestedField schemaField =
schema.findField(fieldId);
+ final Type schemaFieldType = schemaField.type();
+ if(schemaField.isRequired()) {
Review Comment:
```suggestion
if (schemaField.isRequired()) {
```
##########
nifi-nar-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/PutIceberg.java:
##########
@@ -108,6 +108,14 @@ public class PutIceberg extends AbstractIcebergProcessor {
.addValidator(StandardValidators.NON_BLANK_VALIDATOR)
.build();
+ static final PropertyDescriptor UNMATCHED_COLUMN_BEHAVIOR = new
PropertyDescriptor.Builder()
+ .name("unmatched-column-behavior")
+ .displayName("Unmatched Column Behavior")
+ .description("If an incoming record does not have a field mapping
for all of the database table's columns, this property specifies how to handle
the situation")
+ .allowableValues(UnmatchedColumnBehavior.class)
+
.defaultValue(UnmatchedColumnBehavior.FAIL_UNMATCHED_COLUMN.getValue())
+ .required(true)
+ .build();
Review Comment:
```suggestion
.build();
```
##########
nifi-nar-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/PutIceberg.java:
##########
@@ -108,6 +108,14 @@ public class PutIceberg extends AbstractIcebergProcessor {
.addValidator(StandardValidators.NON_BLANK_VALIDATOR)
.build();
+ static final PropertyDescriptor UNMATCHED_COLUMN_BEHAVIOR = new
PropertyDescriptor.Builder()
+ .name("unmatched-column-behavior")
+ .displayName("Unmatched Column Behavior")
+ .description("If an incoming record does not have a field mapping
for all of the database table's columns, this property specifies how to handle
the situation")
Review Comment:
```suggestion
.description("If an incoming record does not have a field
mapping for all of the database table's columns, this property specifies how to
handle the situation.")
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
