diff --git a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroFieldMatcher.java b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroFieldMatcher.java new file mode 100644 index 00000000000000..91e1ad82fa3106 --- /dev/null +++ b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroFieldMatcher.java @@ -0,0 +1,348 @@ +/* + * 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.flink.formats.avro; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.types.logical.RowType; + +import org.apache.avro.AvroRuntimeException; +import org.apache.avro.Schema; +import org.apache.avro.generic.GenericData; +import org.apache.avro.generic.IndexedRecord; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.stream.Collectors; + +/** + * Resolves {@link FieldMatching#NAME} pairings between the fields of a Flink {@link RowType} and + * the fields of an Avro record {@link Schema}. + * + *

Matching is attempted in three stages, so that the more surprising rules can only ever apply + * to fields the stricter rules left over: + * + *

    + *
  1. an exact, case-sensitive comparison of the Avro field name; + *
  2. an exact, case-sensitive comparison against the Avro field aliases; + *
  3. a case-insensitive ({@link Locale#ROOT}) comparison against names and aliases. + *
+ * + *

Anything that cannot be resolved unambiguously, and anything that would silently lose or + * fabricate data, is rejected with a message that names the offending field. In particular a column + * that has no Avro counterpart is an error when writing (the column would be dropped) and an error + * when reading if the column is {@code NOT NULL} (the column could only be read as {@code NULL}). + * Avro fields that no column maps to are tolerated: when writing they must be nullable or declare a + * default, when reading they are simply ignored. + * + *

Resolution is deliberately expensive and done once per (row type, schema) pair; the resulting + * {@link Plan} reduces the per-record cost to an array lookup. + */ +@Internal +public final class AvroFieldMatcher { + + /** Returned by {@link Plan#avroPositionOf(int)} for a column with no Avro counterpart. */ + public static final int UNMATCHED = -1; + + /** Internal marker for a lookup key that would match more than one Avro field. */ + private static final int AMBIGUOUS = -2; + + private AvroFieldMatcher() {} + + /** + * Resolves the pairing used to convert a {@link RowType} into a record of the given schema. + * + * @throws IllegalArgumentException if the pairing is ambiguous, or if it would drop a column, + * or if it would leave an Avro field that is neither nullable nor defaulted unwritten. + */ + public static Plan forSerialization(RowType rowType, Schema recordSchema) { + return resolve(rowType, recordSchema, true); + } + + /** + * Resolves the pairing used to convert a record of the given schema into a {@link RowType}. + * + * @throws IllegalArgumentException if the pairing is ambiguous, or if a {@code NOT NULL} column + * has no Avro counterpart. + */ + public static Plan forDeserialization(RowType rowType, Schema recordSchema) { + return resolve(rowType, recordSchema, false); + } + + private static Plan resolve(RowType rowType, Schema recordSchema, boolean forSerialization) { + if (recordSchema.getType() != Schema.Type.RECORD) { + throw new IllegalArgumentException( + String.format( + "Matching fields by name requires an Avro RECORD schema for row type %s, but got: %s", + rowType, recordSchema)); + } + + final List avroFields = recordSchema.getFields(); + final List rowFieldNames = rowType.getFieldNames(); + final int arity = rowFieldNames.size(); + + final int[] rowToAvroPos = new int[arity]; + final int[] avroPosToRowField = new int[avroFields.size()]; + Arrays.fill(rowToAvroPos, UNMATCHED); + Arrays.fill(avroPosToRowField, UNMATCHED); + + // Stage 1: exact names. Row field names are unique, so this stage cannot conflict with + // itself, and running it to completion first guarantees that a fuzzy match can never + // claim an Avro field that some other column matches exactly. + for (int i = 0; i < arity; i++) { + final Schema.Field avroField = recordSchema.getField(rowFieldNames.get(i)); + if (avroField != null) { + rowToAvroPos[i] = avroField.pos(); + avroPosToRowField[avroField.pos()] = i; + } + } + + // Stages 2 and 3: aliases, then a case-insensitive comparison. + final Map byAlias = new HashMap<>(); + final Map byLowerCase = new HashMap<>(); + for (Schema.Field avroField : avroFields) { + index(byLowerCase, avroField.name(), avroField.pos()); + for (String alias : avroField.aliases()) { + byAlias.merge(alias, avroField.pos(), AvroFieldMatcher::mergePositions); + index(byLowerCase, alias, avroField.pos()); + } + } + + for (int i = 0; i < arity; i++) { + if (rowToAvroPos[i] != UNMATCHED) { + continue; + } + final String rowFieldName = rowFieldNames.get(i); + + String rule = "an Avro field alias"; + int candidate = byAlias.getOrDefault(rowFieldName, UNMATCHED); + if (candidate == UNMATCHED) { + rule = "a case-insensitive comparison"; + candidate = + byLowerCase.getOrDefault(rowFieldName.toLowerCase(Locale.ROOT), UNMATCHED); + } + + if (candidate == UNMATCHED) { + continue; + } + if (candidate == AMBIGUOUS) { + throw new IllegalArgumentException( + String.format( + "Column '%s' matches more than one field of the Avro record '%s' by %s. " + + "Rename the column so that it matches exactly one Avro field " + + "(fields: %s).", + rowFieldName, + recordSchema.getFullName(), + rule, + fieldNames(avroFields))); + } + final int competitor = avroPosToRowField[candidate]; + if (competitor != UNMATCHED) { + throw new IllegalArgumentException( + String.format( + "Columns '%s' and '%s' both match field '%s' of the Avro record '%s'. " + + "Rename one of the columns so that every Avro field is claimed " + + "at most once.", + rowFieldNames.get(competitor), + rowFieldName, + avroFields.get(candidate).name(), + recordSchema.getFullName())); + } + + rowToAvroPos[i] = candidate; + avroPosToRowField[candidate] = i; + } + + return forSerialization + ? serializationPlan(recordSchema, rowToAvroPos, avroPosToRowField, rowFieldNames) + : deserializationPlan(rowType, recordSchema, rowToAvroPos, rowFieldNames); + } + + private static Plan serializationPlan( + Schema recordSchema, + int[] rowToAvroPos, + int[] avroPosToRowField, + List rowFieldNames) { + final List avroFields = recordSchema.getFields(); + + // A column with no Avro counterpart would be dropped without a trace. + for (int i = 0; i < rowToAvroPos.length; i++) { + if (rowToAvroPos[i] == UNMATCHED) { + throw new IllegalArgumentException( + String.format( + "Column '%s' cannot be written: the Avro record '%s' has no field " + + "matching that name (fields: %s). Add the field to the Avro " + + "schema, or project the column away before writing.", + rowFieldNames.get(i), + recordSchema.getFullName(), + fieldNames(avroFields))); + } + } + + // Avro fields that nothing writes to have to be either nullable or defaulted, otherwise + // the record cannot be encoded at all and Avro reports it as an opaque + // NullPointerException at the first record. + final List defaultedPositions = new ArrayList<>(0); + final List defaultValues = new ArrayList<>(0); + for (Schema.Field avroField : avroFields) { + if (avroPosToRowField[avroField.pos()] != UNMATCHED) { + continue; + } + if (avroField.hasDefaultValue()) { + final Object defaultValue = materializeDefault(recordSchema, avroField); + // A null default needs no action: a fresh record is null everywhere. + if (defaultValue != null) { + defaultedPositions.add(avroField.pos()); + defaultValues.add(defaultValue); + } + } else if (!isNullable(avroField.schema())) { + throw new IllegalArgumentException( + String.format( + "Field '%s' of the Avro record '%s' is neither nullable nor does it " + + "declare a default value, but no column matches it " + + "(columns: %s). Add a matching column, or make the Avro field " + + "nullable, or give it a default value.", + avroField.name(), + recordSchema.getFullName(), + String.join(", ", rowFieldNames))); + } + } + + return new Plan( + recordSchema, + rowToAvroPos, + defaultedPositions.stream().mapToInt(Integer::intValue).toArray(), + defaultValues.toArray()); + } + + private static Plan deserializationPlan( + RowType rowType, Schema recordSchema, int[] rowToAvroPos, List rowFieldNames) { + // A NOT NULL column with no Avro counterpart could only ever be read as null, which + // silently violates the contract the rest of the plan relies on. + for (int i = 0; i < rowToAvroPos.length; i++) { + if (rowToAvroPos[i] == UNMATCHED && !rowType.getTypeAt(i).isNullable()) { + throw new IllegalArgumentException( + String.format( + "Column '%s' is declared NOT NULL, but the Avro record '%s' has no " + + "field matching that name (fields: %s), so it could only be " + + "read as NULL. Add the field to the Avro schema, or make the " + + "column nullable.", + rowFieldNames.get(i), + recordSchema.getFullName(), + fieldNames(recordSchema.getFields()))); + } + } + return new Plan(recordSchema, rowToAvroPos, new int[0], new Object[0]); + } + + private static void index(Map lookup, String name, int position) { + lookup.merge(name.toLowerCase(Locale.ROOT), position, AvroFieldMatcher::mergePositions); + } + + private static int mergePositions(int existing, int added) { + return existing == added ? existing : AMBIGUOUS; + } + + private static Object materializeDefault(Schema recordSchema, Schema.Field avroField) { + final GenericData genericData = GenericData.get(); + try { + // Copy the value: GenericData caches one shared instance of every default, and for + // records, arrays and maps that instance would otherwise be aliased by every record + // this converter produces. + return genericData.deepCopy(avroField.schema(), genericData.getDefaultValue(avroField)); + } catch (AvroRuntimeException e) { + throw new IllegalArgumentException( + String.format( + "Cannot use the default value of field '%s' of the Avro record '%s', " + + "which is required because no column matches that field.", + avroField.name(), recordSchema.getFullName()), + e); + } + } + + private static boolean isNullable(Schema schema) { + if (schema.getType() == Schema.Type.NULL) { + return true; + } + return schema.getType() == Schema.Type.UNION + && schema.getTypes().stream().anyMatch(t -> t.getType() == Schema.Type.NULL); + } + + private static String fieldNames(List avroFields) { + return avroFields.stream().map(Schema.Field::name).collect(Collectors.joining(", ")); + } + + /** + * An immutable, resolved pairing between a {@link RowType} and one Avro record {@link Schema}. + * + *

Every field is {@code final} and no reachable state is mutated after construction, so an + * instance may be published through a plain, non-volatile field: the JMM guarantees that a + * thread which observes the reference also observes the arrays it was built with. + */ + public static final class Plan { + + private final Schema recordSchema; + private final int[] rowToAvroPos; + private final int[] defaultedAvroPos; + private final Object[] defaultValues; + + private Plan( + Schema recordSchema, + int[] rowToAvroPos, + int[] defaultedAvroPos, + Object[] defaultValues) { + this.recordSchema = recordSchema; + this.rowToAvroPos = rowToAvroPos; + this.defaultedAvroPos = defaultedAvroPos; + this.defaultValues = defaultValues; + } + + /** + * Whether this plan was resolved against exactly the given schema instance. + * + *

Compares by identity on purpose: it is a cache guard, and Avro's {@code equals} walks + * the whole schema. A miss only costs a re-resolution, and in practice the same instance is + * handed to a converter for its entire lifetime. + */ + public boolean appliesTo(Schema schema) { + return recordSchema == schema; + } + + /** + * The position of the Avro field paired with the given row field, or {@link #UNMATCHED} if + * the row field has no counterpart. + */ + public int avroPositionOf(int rowFieldIndex) { + return rowToAvroPos[rowFieldIndex]; + } + + /** + * Writes the declared default of every Avro field that no row field maps to. Must be called + * once per freshly created record; usually a no-op. + */ + public void fillDefaults(IndexedRecord record) { + for (int i = 0; i < defaultedAvroPos.length; i++) { + record.put(defaultedAvroPos[i], defaultValues[i]); + } + } + } +} diff --git a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroToRowDataConverters.java b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroToRowDataConverters.java index 9c63c56c4dcba7..3b5327cd6ffcb4 100644 --- a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroToRowDataConverters.java +++ b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroToRowDataConverters.java @@ -33,10 +33,13 @@ import org.apache.flink.table.types.logical.RowType; import org.apache.flink.table.types.logical.utils.LogicalTypeUtils; +import org.apache.avro.Schema; import org.apache.avro.generic.GenericFixed; import org.apache.avro.generic.GenericRecord; import org.apache.avro.generic.IndexedRecord; +import javax.annotation.Nullable; + import java.io.Serializable; import java.lang.reflect.Array; import java.nio.ByteBuffer; @@ -74,12 +77,75 @@ public static AvroToRowDataConverter createRowConverter(RowType rowType) { public static AvroToRowDataConverter createRowConverter( RowType rowType, boolean legacyTimestampMapping) { - final AvroToRowDataConverter[] fieldConverters = - rowType.getFields().stream() - .map(RowType.RowField::getType) - .map(type -> createNullableConverter(type, legacyTimestampMapping)) - .toArray(AvroToRowDataConverter[]::new); + return createRowConverterInternal( + rowType, legacyTimestampMapping, null, FieldMatching.INDEX); + } + + /** + * Creates a runtime converter that maps Avro records onto the given row type using the given + * field matching strategy. + * + *

Prefer {@link #createRowConverter(Schema, RowType, FieldMatching)} where the reader schema + * is known: passing it lets a field mismatch be reported when the converter is created rather + * than when the first record arrives. + */ + public static AvroToRowDataConverter createRowConverter( + RowType rowType, FieldMatching fieldMatching) { + return createRowConverterInternal(rowType, true, null, fieldMatching); + } + + /** + * Creates a runtime converter that maps records of the given reader schema onto the given row + * type using the given field matching strategy. + */ + public static AvroToRowDataConverter createRowConverter( + Schema readerSchema, RowType rowType, FieldMatching fieldMatching) { + return createRowConverter(readerSchema, rowType, true, fieldMatching); + } + + /** + * Creates a runtime converter that maps records of the given reader schema onto the given row + * type using the given field matching strategy. + */ + public static AvroToRowDataConverter createRowConverter( + Schema readerSchema, + RowType rowType, + boolean legacyTimestampMapping, + FieldMatching fieldMatching) { + return createRowConverterInternal( + rowType, legacyTimestampMapping, readerSchema, fieldMatching); + } + + private static AvroToRowDataConverter createRowConverterInternal( + RowType rowType, + boolean legacyTimestampMapping, + @Nullable Schema schema, + FieldMatching fieldMatching) { final int arity = rowType.getFieldCount(); + final Schema recordSchema = asRecordSchema(schema); + + // Resolving here is not what the runtime converter uses - the reader schema instance seen + // at runtime is a different one, because the schema travels to the task as a string. It + // serves two other purposes: reporting a field mismatch while the job is still being + // assembled, and telling each nested converter which Avro field it is going to read. + final AvroFieldMatcher.Plan plan = + fieldMatching == FieldMatching.NAME && recordSchema != null + ? AvroFieldMatcher.forDeserialization(rowType, recordSchema) + : null; + + final AvroToRowDataConverter[] fieldConverters = new AvroToRowDataConverter[arity]; + for (int i = 0; i < arity; i++) { + fieldConverters[i] = + createNullableConverter( + rowType.getTypeAt(i), + legacyTimestampMapping, + avroFieldSchema(recordSchema, plan, i), + fieldMatching); + } + + if (fieldMatching == FieldMatching.NAME) { + return new NameMatchingRowConverter(rowType, fieldConverters); + } return avroObject -> { IndexedRecord record = (IndexedRecord) avroObject; @@ -93,10 +159,65 @@ public static AvroToRowDataConverter createRowConverter( }; } + /** + * Reads Avro record fields into the row field of the same name, so that the two may declare + * their fields in a different order. See {@link AvroFieldMatcher} for the matching rules. + */ + private static final class NameMatchingRowConverter implements AvroToRowDataConverter { + + private static final long serialVersionUID = 1L; + + private final RowType rowType; + private final AvroToRowDataConverter[] fieldConverters; + + /** + * The pairing resolved for the schema seen last. Records of a stream all share one reader + * schema instance, so this is effectively resolved once. The plan is immutable and safely + * publishable, hence no synchronization: the worst a racy read can cost is one redundant + * resolution. + */ + private transient AvroFieldMatcher.Plan plan; + + private NameMatchingRowConverter( + RowType rowType, AvroToRowDataConverter[] fieldConverters) { + this.rowType = rowType; + this.fieldConverters = fieldConverters; + } + + @Override + public Object convert(Object avroObject) { + final IndexedRecord record = (IndexedRecord) avroObject; + final Schema recordSchema = record.getSchema(); + + AvroFieldMatcher.Plan currentPlan = plan; + if (currentPlan == null || !currentPlan.appliesTo(recordSchema)) { + currentPlan = AvroFieldMatcher.forDeserialization(rowType, recordSchema); + plan = currentPlan; + } + + final GenericRowData row = new GenericRowData(fieldConverters.length); + for (int i = 0; i < fieldConverters.length; ++i) { + final int avroPos = currentPlan.avroPositionOf(i); + // A column the record has no field for stays null; the plan already refused the + // case where that would violate a NOT NULL column. + if (avroPos != AvroFieldMatcher.UNMATCHED) { + // avro always deserialize successfully even though the type isn't matched + // so no need to throw exception about which field can't be deserialized + row.setField(i, fieldConverters[i].convert(record.get(avroPos))); + } + } + return row; + } + } + /** Creates a runtime converter which is null safe. */ private static AvroToRowDataConverter createNullableConverter( - LogicalType type, boolean legacyTimestampMapping) { - final AvroToRowDataConverter converter = createConverter(type, legacyTimestampMapping); + LogicalType type, + boolean legacyTimestampMapping, + @Nullable Schema schema, + FieldMatching fieldMatching) { + final AvroToRowDataConverter converter = + createConverter(type, legacyTimestampMapping, schema, fieldMatching); return avroObject -> { if (avroObject == null) { return null; @@ -107,7 +228,10 @@ private static AvroToRowDataConverter createNullableConverter( /** Creates a runtime converter which assuming input object is not null. */ private static AvroToRowDataConverter createConverter( - LogicalType type, boolean legacyTimestampMapping) { + LogicalType type, + boolean legacyTimestampMapping, + @Nullable Schema schema, + FieldMatching fieldMatching) { switch (type.getTypeRoot()) { case NULL: return avroObject -> null; @@ -144,12 +268,18 @@ private static AvroToRowDataConverter createConverter( case DECIMAL: return createDecimalConverter((DecimalType) type); case ARRAY: - return createArrayConverter((ArrayType) type, legacyTimestampMapping); + return createArrayConverter( + (ArrayType) type, + legacyTimestampMapping, + elementSchemaOf(schema), + fieldMatching); case ROW: - return createRowConverter((RowType) type); + return createRowConverterInternal( + (RowType) type, legacyTimestampMapping, schema, fieldMatching); case MAP: case MULTISET: - return createMapConverter(type, legacyTimestampMapping); + return createMapConverter( + type, legacyTimestampMapping, valueSchemaOf(schema), fieldMatching); case RAW: default: throw new UnsupportedOperationException("Unsupported type: " + type); @@ -175,9 +305,16 @@ private static AvroToRowDataConverter createDecimalConverter(DecimalType decimal } private static AvroToRowDataConverter createArrayConverter( - ArrayType arrayType, boolean legacyTimestampMapping) { + ArrayType arrayType, + boolean legacyTimestampMapping, + @Nullable Schema elementSchema, + FieldMatching fieldMatching) { final AvroToRowDataConverter elementConverter = - createNullableConverter(arrayType.getElementType(), legacyTimestampMapping); + createNullableConverter( + arrayType.getElementType(), + legacyTimestampMapping, + elementSchema, + fieldMatching); final Class elementClass = LogicalTypeUtils.toInternalConversionClass(arrayType.getElementType()); @@ -193,11 +330,22 @@ private static AvroToRowDataConverter createArrayConverter( } private static AvroToRowDataConverter createMapConverter( - LogicalType type, boolean legacyTimestampMapping) { + LogicalType type, + boolean legacyTimestampMapping, + @Nullable Schema valueSchema, + FieldMatching fieldMatching) { final AvroToRowDataConverter keyConverter = - createConverter(DataTypes.STRING().getLogicalType(), legacyTimestampMapping); + createConverter( + DataTypes.STRING().getLogicalType(), + legacyTimestampMapping, + null, + fieldMatching); final AvroToRowDataConverter valueConverter = - createNullableConverter(extractValueTypeToAvroMap(type), legacyTimestampMapping); + createNullableConverter( + extractValueTypeToAvroMap(type), + legacyTimestampMapping, + valueSchema, + fieldMatching); return avroObject -> { final Map map = (Map) avroObject; @@ -211,6 +359,68 @@ private static AvroToRowDataConverter createMapConverter( }; } + // ------------------------------------------------------------------------------------- + // Reader schema navigation + // + // The reader schema is optional throughout: it is only used to validate eagerly and to hand + // each nested converter its own schema. Wherever a schema cannot be narrowed down to a single + // Avro type - a true multi-type union, say - navigation yields null and the converters fall + // back to resolving from the record they are handed at runtime. + // ------------------------------------------------------------------------------------- + + private static @Nullable Schema avroFieldSchema( + @Nullable Schema recordSchema, + @Nullable AvroFieldMatcher.Plan plan, + int rowFieldIndex) { + if (recordSchema == null) { + return null; + } + final int position = plan == null ? rowFieldIndex : plan.avroPositionOf(rowFieldIndex); + final List fields = recordSchema.getFields(); + if (position == AvroFieldMatcher.UNMATCHED || position >= fields.size()) { + return null; + } + return fields.get(position).schema(); + } + + private static @Nullable Schema asRecordSchema(@Nullable Schema schema) { + final Schema resolved = unwrapNullableUnion(schema); + return resolved != null && resolved.getType() == Schema.Type.RECORD ? resolved : null; + } + + private static @Nullable Schema elementSchemaOf(@Nullable Schema schema) { + final Schema resolved = unwrapNullableUnion(schema); + return resolved != null && resolved.getType() == Schema.Type.ARRAY + ? resolved.getElementType() + : null; + } + + private static @Nullable Schema valueSchemaOf(@Nullable Schema schema) { + final Schema resolved = unwrapNullableUnion(schema); + return resolved != null && resolved.getType() == Schema.Type.MAP + ? resolved.getValueType() + : null; + } + + private static @Nullable Schema unwrapNullableUnion(@Nullable Schema schema) { + if (schema == null || schema.getType() != Schema.Type.UNION) { + return schema; + } + Schema resolved = null; + for (Schema branch : schema.getTypes()) { + if (branch.getType() == Schema.Type.NULL) { + continue; + } + if (resolved != null) { + // More than one non-null branch: the reader schema of a field cannot be pinned + // down statically, so give up rather than guess. + return null; + } + resolved = branch; + } + return resolved; + } + private static TimestampData convertToTimestamp(Object object) { final long millis; if (object instanceof Long) { diff --git a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/FieldMatching.java b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/FieldMatching.java new file mode 100644 index 00000000000000..e58cd31c8df636 --- /dev/null +++ b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/FieldMatching.java @@ -0,0 +1,69 @@ +/* + * 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.flink.formats.avro; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.configuration.DescribedEnum; +import org.apache.flink.configuration.description.InlineElement; +import org.apache.flink.table.types.logical.RowType; + +import static org.apache.flink.configuration.description.TextElement.text; + +/** + * Strategy used to pair the fields of a Flink {@link RowType} with the fields of an Avro record + * schema. + * + *

{@link #INDEX} is the historical - and still the default - behaviour. It is the right choice + * whenever the Avro schema is derived from the row type, because then both sides are guaranteed to + * agree on field order. {@link #NAME} exists for the case where the Avro schema is supplied + * independently of the table schema, for instance by a schema registry, where the two orders need + * not agree. + */ +@PublicEvolving +public enum FieldMatching implements DescribedEnum { + INDEX( + "index", + text( + "Pair the n-th row field with the n-th Avro field. Requires both schemas to " + + "declare their fields in the same order.")), + NAME( + "name", + text( + "Pair fields by name, so that field order may differ between the row type and " + + "the Avro schema. Names are compared exactly first, then against Avro " + + "field aliases, and finally ignoring case.")); + + private final String value; + private final InlineElement description; + + FieldMatching(String value, InlineElement description) { + this.value = value; + this.description = description; + } + + @Override + public String toString() { + return value; + } + + @Override + public InlineElement getDescription() { + return description; + } +} diff --git a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/RowDataToAvroConverters.java b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/RowDataToAvroConverters.java index 202cd601bc02eb..e9f8ea18520b69 100644 --- a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/RowDataToAvroConverters.java +++ b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/RowDataToAvroConverters.java @@ -77,6 +77,21 @@ public static RowDataToAvroConverter createConverter(LogicalType type) { public static RowDataToAvroConverter createConverter( LogicalType type, boolean legacyTimestampMapping) { + return createConverter(type, legacyTimestampMapping, FieldMatching.INDEX); + } + + /** + * Creates a runtime converter according to the given logical type that converts objects of + * Flink Table & SQL internal data structures to corresponding Avro data structures. + * + * @param legacyTimestampMapping whether to use the legacy mapping of Flink timestamp types onto + * Avro timestamp types. + * @param fieldMatching how the fields of a {@link RowType} are paired with the fields of the + * target Avro record schema. Only {@link FieldMatching#NAME} tolerates the two declaring + * their fields in a different order. + */ + public static RowDataToAvroConverter createConverter( + LogicalType type, boolean legacyTimestampMapping, FieldMatching fieldMatching) { final RowDataToAvroConverter converter; switch (type.getTypeRoot()) { case NULL: @@ -211,14 +226,17 @@ public Object convert(Schema schema, Object object) { }; break; case ARRAY: - converter = createArrayConverter((ArrayType) type, legacyTimestampMapping); + converter = + createArrayConverter( + (ArrayType) type, legacyTimestampMapping, fieldMatching); break; case ROW: - converter = createRowConverter((RowType) type, legacyTimestampMapping); + converter = + createRowConverter((RowType) type, legacyTimestampMapping, fieldMatching); break; case MAP: case MULTISET: - converter = createMapConverter(type, legacyTimestampMapping); + converter = createMapConverter(type, legacyTimestampMapping, fieldMatching); break; case RAW: default: @@ -257,10 +275,13 @@ public Object convert(Schema schema, Object object) { } private static RowDataToAvroConverter createRowConverter( - RowType rowType, boolean legacyTimestampMapping) { + RowType rowType, boolean legacyTimestampMapping, FieldMatching fieldMatching) { final RowDataToAvroConverter[] fieldConverters = rowType.getChildren().stream() - .map(legacyType -> createConverter(legacyType, legacyTimestampMapping)) + .map( + fieldType -> + createConverter( + fieldType, legacyTimestampMapping, fieldMatching)) .toArray(RowDataToAvroConverter[]::new); final LogicalType[] fieldTypes = rowType.getFields().stream() @@ -270,41 +291,116 @@ private static RowDataToAvroConverter createRowConverter( for (int i = 0; i < fieldTypes.length; i++) { fieldGetters[i] = RowData.createFieldGetter(fieldTypes[i], i); } - final int length = rowType.getFieldCount(); - return new RowDataToAvroConverter() { - private static final long serialVersionUID = 1L; + return fieldMatching == FieldMatching.NAME + ? new NameMatchingRowConverter(rowType, fieldConverters, fieldGetters) + : new IndexMatchingRowConverter(fieldConverters, fieldGetters); + } - @Override - public Object convert(Schema schema, Object object) { - final RowData row = (RowData) object; - final List fields = schema.getFields(); - final GenericRecord record = new GenericData.Record(schema); - for (int i = 0; i < length; ++i) { - final Schema.Field schemaField = fields.get(i); - try { - Object avroObject = - fieldConverters[i].convert( - schemaField.schema(), fieldGetters[i].getFieldOrNull(row)); - record.put(i, avroObject); - } catch (Throwable t) { - throw new RuntimeException( - String.format( - "Fail to serialize at field: %s.", schemaField.name()), - t); - } + /** Pairs the n-th field of the row with the n-th field of the Avro record schema. */ + private static final class IndexMatchingRowConverter implements RowDataToAvroConverter { + + private static final long serialVersionUID = 1L; + + private final RowDataToAvroConverter[] fieldConverters; + private final RowData.FieldGetter[] fieldGetters; + + private IndexMatchingRowConverter( + RowDataToAvroConverter[] fieldConverters, RowData.FieldGetter[] fieldGetters) { + this.fieldConverters = fieldConverters; + this.fieldGetters = fieldGetters; + } + + @Override + public Object convert(Schema schema, Object object) { + final RowData row = (RowData) object; + final List fields = schema.getFields(); + final GenericRecord record = new GenericData.Record(schema); + for (int i = 0; i < fieldConverters.length; ++i) { + final Schema.Field schemaField = fields.get(i); + try { + Object avroObject = + fieldConverters[i].convert( + schemaField.schema(), fieldGetters[i].getFieldOrNull(row)); + record.put(i, avroObject); + } catch (Throwable t) { + throw new RuntimeException( + String.format("Fail to serialize at field: %s.", schemaField.name()), + t); } - return record; } - }; + return record; + } + } + + /** + * Pairs row fields with Avro record fields by name, so that the two may declare their fields in + * a different order. See {@link AvroFieldMatcher} for the matching rules. + */ + private static final class NameMatchingRowConverter implements RowDataToAvroConverter { + + private static final long serialVersionUID = 1L; + + private final RowType rowType; + private final RowDataToAvroConverter[] fieldConverters; + private final RowData.FieldGetter[] fieldGetters; + + /** + * The pairing resolved for the schema seen last. Callers hand the same schema instance to + * every {@link #convert} call, so this is effectively resolved once. The plan is immutable + * and safely publishable, hence no synchronization: the worst a racy read can cost is one + * redundant resolution. + */ + private transient AvroFieldMatcher.Plan plan; + + private NameMatchingRowConverter( + RowType rowType, + RowDataToAvroConverter[] fieldConverters, + RowData.FieldGetter[] fieldGetters) { + this.rowType = rowType; + this.fieldConverters = fieldConverters; + this.fieldGetters = fieldGetters; + } + + @Override + public Object convert(Schema schema, Object object) { + AvroFieldMatcher.Plan currentPlan = plan; + if (currentPlan == null || !currentPlan.appliesTo(schema)) { + currentPlan = AvroFieldMatcher.forSerialization(rowType, schema); + plan = currentPlan; + } + + final RowData row = (RowData) object; + final List fields = schema.getFields(); + final GenericRecord record = new GenericData.Record(schema); + // Avro fields no column maps to keep their declared default rather than staying null. + currentPlan.fillDefaults(record); + + for (int i = 0; i < fieldConverters.length; ++i) { + // A serialization plan never leaves a column unmatched, so this is always a field. + final int avroPos = currentPlan.avroPositionOf(i); + final Schema.Field schemaField = fields.get(avroPos); + try { + Object avroObject = + fieldConverters[i].convert( + schemaField.schema(), fieldGetters[i].getFieldOrNull(row)); + record.put(avroPos, avroObject); + } catch (Throwable t) { + throw new RuntimeException( + String.format("Fail to serialize at field: %s.", schemaField.name()), + t); + } + } + return record; + } } private static RowDataToAvroConverter createArrayConverter( - ArrayType arrayType, boolean legacyTimestampMapping) { + ArrayType arrayType, boolean legacyTimestampMapping, FieldMatching fieldMatching) { LogicalType elementType = arrayType.getElementType(); final ArrayData.ElementGetter elementGetter = ArrayData.createElementGetter(elementType); final RowDataToAvroConverter elementConverter = - createConverter(arrayType.getElementType(), legacyTimestampMapping); + createConverter(elementType, legacyTimestampMapping, fieldMatching); return new RowDataToAvroConverter() { private static final long serialVersionUID = 1L; @@ -325,11 +421,11 @@ public Object convert(Schema schema, Object object) { } private static RowDataToAvroConverter createMapConverter( - LogicalType type, boolean legacyTimestampMapping) { + LogicalType type, boolean legacyTimestampMapping, FieldMatching fieldMatching) { LogicalType valueType = extractValueTypeToAvroMap(type); final ArrayData.ElementGetter valueGetter = ArrayData.createElementGetter(valueType); final RowDataToAvroConverter valueConverter = - createConverter(valueType, legacyTimestampMapping); + createConverter(valueType, legacyTimestampMapping, fieldMatching); return new RowDataToAvroConverter() { private static final long serialVersionUID = 1L; diff --git a/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroFieldMatcherTest.java b/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroFieldMatcherTest.java new file mode 100644 index 00000000000000..8fb030890002b0 --- /dev/null +++ b/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroFieldMatcherTest.java @@ -0,0 +1,217 @@ +/* + * 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.flink.formats.avro; + +import org.apache.flink.table.types.logical.RowType; + +import org.apache.avro.Schema; +import org.apache.avro.generic.GenericData; +import org.apache.avro.generic.GenericRecord; +import org.junit.jupiter.api.Test; + +import static org.apache.flink.table.api.DataTypes.BOOLEAN; +import static org.apache.flink.table.api.DataTypes.FIELD; +import static org.apache.flink.table.api.DataTypes.INT; +import static org.apache.flink.table.api.DataTypes.ROW; +import static org.apache.flink.table.api.DataTypes.STRING; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for {@link AvroFieldMatcher}. */ +class AvroFieldMatcherTest { + + @Test + void testMatchesReorderedFields() { + final RowType rowType = rowType(FIELD("a", STRING()), FIELD("b", INT())); + final Schema schema = parse("{'name':'b','type':'int'}, {'name':'a','type':'string'}"); + + final AvroFieldMatcher.Plan plan = AvroFieldMatcher.forSerialization(rowType, schema); + + assertThat(plan.avroPositionOf(0)).isEqualTo(1); + assertThat(plan.avroPositionOf(1)).isZero(); + } + + @Test + void testMatchesIgnoringCase() { + final RowType rowType = rowType(FIELD("firstName", STRING())); + final Schema schema = parse("{'name':'FIRSTNAME','type':'string'}"); + + assertThat(AvroFieldMatcher.forSerialization(rowType, schema).avroPositionOf(0)).isZero(); + } + + @Test + void testMatchesAvroFieldAlias() { + final RowType rowType = rowType(FIELD("legacy_name", STRING())); + final Schema schema = parse("{'name':'name','aliases':['legacy_name'],'type':'string'}"); + + assertThat(AvroFieldMatcher.forSerialization(rowType, schema).avroPositionOf(0)).isZero(); + } + + @Test + void testExactMatchWinsOverCaseInsensitiveMatch() { + // Both columns match both fields when ignoring case, so only running the exact stage to + // completion first can pair them up the way the user wrote them. + final RowType rowType = rowType(FIELD("foo", STRING()), FIELD("FOO", STRING())); + final Schema schema = + parse("{'name':'FOO','type':'string'}, {'name':'foo','type':'string'}"); + + final AvroFieldMatcher.Plan plan = AvroFieldMatcher.forSerialization(rowType, schema); + + assertThat(plan.avroPositionOf(0)).isEqualTo(1); + assertThat(plan.avroPositionOf(1)).isZero(); + } + + @Test + void testAmbiguousCaseInsensitiveMatchIsRejected() { + final RowType rowType = rowType(FIELD("Foo", STRING())); + final Schema schema = + parse("{'name':'foo','type':'string'}, {'name':'FOO','type':'string'}"); + + assertThatThrownBy(() -> AvroFieldMatcher.forSerialization(rowType, schema)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Column 'Foo' matches more than one field") + .hasMessageContaining("a case-insensitive comparison"); + } + + @Test + void testTwoColumnsCompetingForOneFieldAreRejected() { + final RowType rowType = rowType(FIELD("Foo", STRING()), FIELD("foo", STRING())); + final Schema schema = parse("{'name':'Foo','type':'string'}"); + + assertThatThrownBy(() -> AvroFieldMatcher.forSerialization(rowType, schema)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Columns 'Foo' and 'foo' both match field 'Foo'"); + } + + @Test + void testSerializationRejectsColumnWithoutAvroField() { + final RowType rowType = rowType(FIELD("a", STRING()), FIELD("missing", INT())); + final Schema schema = parse("{'name':'a','type':'string'}"); + + assertThatThrownBy(() -> AvroFieldMatcher.forSerialization(rowType, schema)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Column 'missing' cannot be written") + .hasMessageContaining("fields: a"); + } + + @Test + void testSerializationRejectsUnwritableRequiredAvroField() { + final RowType rowType = rowType(FIELD("a", STRING())); + final Schema schema = + parse("{'name':'a','type':'string'}, {'name':'required','type':'int'}"); + + assertThatThrownBy(() -> AvroFieldMatcher.forSerialization(rowType, schema)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Field 'required'") + .hasMessageContaining("neither nullable nor does it declare a default value"); + } + + @Test + void testSerializationToleratesUnmatchedNullableAvroField() { + final RowType rowType = rowType(FIELD("a", STRING())); + final Schema schema = + parse("{'name':'a','type':'string'}, {'name':'spare','type':['null','int']}"); + + final AvroFieldMatcher.Plan plan = AvroFieldMatcher.forSerialization(rowType, schema); + + final GenericRecord record = new GenericData.Record(schema); + plan.fillDefaults(record); + assertThat(record.get("spare")).isNull(); + } + + @Test + void testSerializationAppliesDefaultOfUnmatchedAvroField() { + final RowType rowType = rowType(FIELD("a", STRING())); + final Schema schema = + parse("{'name':'a','type':'string'}, {'name':'version','type':'int','default':7}"); + + final AvroFieldMatcher.Plan plan = AvroFieldMatcher.forSerialization(rowType, schema); + + final GenericRecord record = new GenericData.Record(schema); + plan.fillDefaults(record); + assertThat(record.get("version")).isEqualTo(7); + } + + @Test + void testDeserializationLeavesNullableColumnUnmatched() { + final RowType rowType = rowType(FIELD("a", STRING()), FIELD("absent", INT())); + final Schema schema = parse("{'name':'a','type':'string'}"); + + final AvroFieldMatcher.Plan plan = AvroFieldMatcher.forDeserialization(rowType, schema); + + assertThat(plan.avroPositionOf(0)).isZero(); + assertThat(plan.avroPositionOf(1)).isEqualTo(AvroFieldMatcher.UNMATCHED); + } + + @Test + void testDeserializationRejectsUnmatchedNotNullColumn() { + final RowType rowType = rowType(FIELD("a", STRING()), FIELD("absent", INT().notNull())); + final Schema schema = parse("{'name':'a','type':'string'}"); + + assertThatThrownBy(() -> AvroFieldMatcher.forDeserialization(rowType, schema)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Column 'absent' is declared NOT NULL") + .hasMessageContaining("could only be read as NULL"); + } + + @Test + void testDeserializationIgnoresUnreadAvroField() { + final RowType rowType = rowType(FIELD("a", STRING())); + final Schema schema = parse("{'name':'a','type':'string'}, {'name':'extra','type':'int'}"); + + assertThat(AvroFieldMatcher.forDeserialization(rowType, schema).avroPositionOf(0)).isZero(); + } + + @Test + void testNonRecordSchemaIsRejected() { + final RowType rowType = rowType(FIELD("a", STRING())); + + assertThatThrownBy( + () -> + AvroFieldMatcher.forSerialization( + rowType, Schema.create(Schema.Type.STRING))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("requires an Avro RECORD schema"); + } + + @Test + void testPlanIsBoundToTheSchemaItWasResolvedFor() { + final RowType rowType = rowType(FIELD("a", STRING()), FIELD("b", BOOLEAN())); + final Schema schema = parse("{'name':'a','type':'string'}, {'name':'b','type':'boolean'}"); + final Schema equalButDistinct = + parse("{'name':'a','type':'string'}, {'name':'b','type':'boolean'}"); + + final AvroFieldMatcher.Plan plan = AvroFieldMatcher.forSerialization(rowType, schema); + + assertThat(plan.appliesTo(schema)).isTrue(); + assertThat(plan.appliesTo(equalButDistinct)).isFalse(); + } + + private static RowType rowType(org.apache.flink.table.api.DataTypes.Field... fields) { + return (RowType) ROW(fields).notNull().getLogicalType(); + } + + /** Parses a record schema from the given field list, using {@code '} instead of {@code "}. */ + private static Schema parse(String fields) { + return new Schema.Parser() + .parse( + ("{'type':'record','name':'TestRecord','fields':[" + fields + "]}") + .replace('\'', '"')); + } +} diff --git a/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroRowDataFieldMatchingTest.java b/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroRowDataFieldMatchingTest.java new file mode 100644 index 00000000000000..4a732496dd7925 --- /dev/null +++ b/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroRowDataFieldMatchingTest.java @@ -0,0 +1,465 @@ +/* + * 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.flink.formats.avro; + +import org.apache.flink.formats.avro.AvroFormatOptions.AvroEncoding; +import org.apache.flink.table.data.GenericArrayData; +import org.apache.flink.table.data.GenericMapData; +import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.data.StringData; +import org.apache.flink.table.runtime.typeutils.InternalTypeInfo; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.util.InstantiationUtil; + +import org.apache.avro.Schema; +import org.apache.avro.generic.GenericData; +import org.apache.avro.generic.GenericDatumWriter; +import org.apache.avro.generic.GenericRecord; +import org.apache.avro.generic.IndexedRecord; +import org.apache.avro.io.Encoder; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; + +import java.io.ByteArrayOutputStream; +import java.util.Collections; +import java.util.List; +import java.util.Map; + +import static org.apache.flink.formats.avro.utils.AvroTestUtils.createEncoder; +import static org.apache.flink.table.api.DataTypes.ARRAY; +import static org.apache.flink.table.api.DataTypes.BOOLEAN; +import static org.apache.flink.table.api.DataTypes.FIELD; +import static org.apache.flink.table.api.DataTypes.INT; +import static org.apache.flink.table.api.DataTypes.MAP; +import static org.apache.flink.table.api.DataTypes.ROW; +import static org.apache.flink.table.api.DataTypes.STRING; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Tests for {@link FieldMatching#NAME}, which pairs the fields of a {@link RowType} with the fields + * of an Avro record schema by name rather than by position. + * + *

All of these need an Avro schema that is supplied independently of the row type: a schema + * derived from the row type by {@link org.apache.flink.formats.avro.typeutils.AvroSchemaConverter} + * always agrees on field order, so there is nothing for name matching to fix. + */ +class AvroRowDataFieldMatchingTest { + + @ParameterizedTest + @EnumSource(AvroEncoding.class) + void testFieldOrderMismatch(AvroEncoding encoding) throws Exception { + final Schema avroSchema = + record( + "{'name':'c','type':'boolean'}", + "{'name':'a','type':'string'}", + "{'name':'b','type':'int'}"); + final RowType rowType = + rowType( + FIELD("a", STRING().notNull()), + FIELD("b", INT().notNull()), + FIELD("c", BOOLEAN().notNull())); + + final byte[] serialized = + serializer(rowType, avroSchema, encoding, FieldMatching.NAME) + .serialize(GenericRowData.of(StringData.fromString("hello"), 42, true)); + + // Byte-for-byte identical to what an Avro writer produces for that schema. + final GenericRecord expected = new GenericData.Record(avroSchema); + expected.put("a", "hello"); + expected.put("b", 42); + expected.put("c", true); + assertThat(serialized).isEqualTo(encode(expected, avroSchema, encoding)); + + final RowData roundTripped = + deserializer(rowType, avroSchema, encoding, FieldMatching.NAME) + .deserialize(serialized); + assertThat(roundTripped.getString(0).toString()).isEqualTo("hello"); + assertThat(roundTripped.getInt(1)).isEqualTo(42); + assertThat(roundTripped.getBoolean(2)).isTrue(); + } + + @ParameterizedTest + @EnumSource(AvroEncoding.class) + void testIndexMatchingIgnoresFieldNames(AvroEncoding encoding) throws Exception { + // Spells out the behaviour FieldMatching.NAME exists to avoid: with index matching, a + // reordered schema of same-typed fields silently swaps the values. + final Schema avroSchema = + record("{'name':'b','type':'string'}", "{'name':'a','type':'string'}"); + final RowType rowType = + rowType(FIELD("a", STRING().notNull()), FIELD("b", STRING().notNull())); + final GenericRowData row = + GenericRowData.of( + StringData.fromString("valueOfA"), StringData.fromString("valueOfB")); + + final GenericRecord viaIndex = + decode( + serializer(rowType, avroSchema, encoding, FieldMatching.INDEX) + .serialize(row), + avroSchema, + encoding); + assertThat(viaIndex.get("b").toString()).isEqualTo("valueOfA"); + assertThat(viaIndex.get("a").toString()).isEqualTo("valueOfB"); + + final GenericRecord viaName = + decode( + serializer(rowType, avroSchema, encoding, FieldMatching.NAME) + .serialize(row), + avroSchema, + encoding); + assertThat(viaName.get("a").toString()).isEqualTo("valueOfA"); + assertThat(viaName.get("b").toString()).isEqualTo("valueOfB"); + } + + @ParameterizedTest + @EnumSource(AvroEncoding.class) + void testCaseInsensitiveFieldMatching(AvroEncoding encoding) throws Exception { + final Schema avroSchema = + record( + "{'name':'LASTNAME','type':'string'}", + "{'name':'firstname','type':'string'}"); + final RowType rowType = + rowType( + FIELD("firstName", STRING().notNull()), + FIELD("lastName", STRING().notNull())); + + final GenericRecord written = + decode( + serializer(rowType, avroSchema, encoding, FieldMatching.NAME) + .serialize( + GenericRowData.of( + StringData.fromString("Ada"), + StringData.fromString("Lovelace"))), + avroSchema, + encoding); + + assertThat(written.get("firstname").toString()).isEqualTo("Ada"); + assertThat(written.get("LASTNAME").toString()).isEqualTo("Lovelace"); + } + + @ParameterizedTest + @EnumSource(AvroEncoding.class) + void testAvroFieldAliasMatching(AvroEncoding encoding) throws Exception { + final Schema avroSchema = record("{'name':'full_name','aliases':['name'],'type':'string'}"); + final RowType rowType = rowType(FIELD("name", STRING().notNull())); + + final GenericRecord written = + decode( + serializer(rowType, avroSchema, encoding, FieldMatching.NAME) + .serialize(GenericRowData.of(StringData.fromString("Grace"))), + avroSchema, + encoding); + + assertThat(written.get("full_name").toString()).isEqualTo("Grace"); + } + + @ParameterizedTest + @EnumSource(AvroEncoding.class) + void testFieldOrderMismatchInNestedRow(AvroEncoding encoding) throws Exception { + final Schema avroSchema = + record( + "{'name':'nested','type':{'type':'record','name':'Nested','fields':[" + + "{'name':'y','type':'int'},{'name':'x','type':'string'}]}}", + "{'name':'id','type':'int'}"); + final RowType rowType = + rowType( + FIELD("id", INT().notNull()), + FIELD( + "nested", + ROW(FIELD("x", STRING().notNull()), FIELD("y", INT().notNull())) + .notNull())); + + final GenericRowData row = + GenericRowData.of(1, GenericRowData.of(StringData.fromString("deep"), 9)); + final byte[] serialized = + serializer(rowType, avroSchema, encoding, FieldMatching.NAME).serialize(row); + + final GenericRecord written = decode(serialized, avroSchema, encoding); + assertThat(written.get("id")).isEqualTo(1); + final GenericRecord writtenNested = (GenericRecord) written.get("nested"); + assertThat(writtenNested.get("x").toString()).isEqualTo("deep"); + assertThat(writtenNested.get("y")).isEqualTo(9); + + final RowData roundTripped = + deserializer(rowType, avroSchema, encoding, FieldMatching.NAME) + .deserialize(serialized); + assertThat(roundTripped.getInt(0)).isEqualTo(1); + assertThat(roundTripped.getRow(1, 2).getString(0).toString()).isEqualTo("deep"); + assertThat(roundTripped.getRow(1, 2).getInt(1)).isEqualTo(9); + } + + @ParameterizedTest + @EnumSource(AvroEncoding.class) + void testFieldOrderMismatchInArrayElement(AvroEncoding encoding) throws Exception { + final Schema avroSchema = + record( + "{'name':'items','type':{'type':'array','items':" + + "{'type':'record','name':'Item','fields':[" + + "{'name':'quantity','type':'int'},{'name':'sku','type':'string'}]}}}"); + final RowType rowType = + rowType( + FIELD( + "items", + ARRAY( + ROW( + FIELD("sku", STRING().notNull()), + FIELD("quantity", INT().notNull())) + .notNull()) + .notNull())); + + final GenericRowData row = + GenericRowData.of( + new GenericArrayData( + new Object[] { + GenericRowData.of(StringData.fromString("A-1"), 3), + GenericRowData.of(StringData.fromString("B-2"), 5) + })); + final byte[] serialized = + serializer(rowType, avroSchema, encoding, FieldMatching.NAME).serialize(row); + + @SuppressWarnings("unchecked") + final List items = + (List) decode(serialized, avroSchema, encoding).get("items"); + assertThat(items).hasSize(2); + assertThat(items.get(0).get("sku").toString()).isEqualTo("A-1"); + assertThat(items.get(0).get("quantity")).isEqualTo(3); + assertThat(items.get(1).get("sku").toString()).isEqualTo("B-2"); + + final RowData roundTripped = + deserializer(rowType, avroSchema, encoding, FieldMatching.NAME) + .deserialize(serialized); + assertThat(roundTripped.getArray(0).getRow(0, 2).getString(0).toString()).isEqualTo("A-1"); + assertThat(roundTripped.getArray(0).getRow(1, 2).getInt(1)).isEqualTo(5); + } + + @ParameterizedTest + @EnumSource(AvroEncoding.class) + void testFieldOrderMismatchInMapValue(AvroEncoding encoding) throws Exception { + final Schema avroSchema = + record( + "{'name':'byKey','type':{'type':'map','values':" + + "{'type':'record','name':'Value','fields':[" + + "{'name':'count','type':'int'},{'name':'label','type':'string'}]}}}"); + final RowType rowType = + rowType( + FIELD( + "byKey", + MAP( + STRING().notNull(), + ROW( + FIELD("label", STRING().notNull()), + FIELD("count", INT().notNull())) + .notNull()) + .notNull())); + + final GenericRowData row = + GenericRowData.of( + new GenericMapData( + Collections.singletonMap( + StringData.fromString("k"), + GenericRowData.of(StringData.fromString("hits"), 11)))); + final byte[] serialized = + serializer(rowType, avroSchema, encoding, FieldMatching.NAME).serialize(row); + + @SuppressWarnings("unchecked") + final Map byKey = + (Map) decode(serialized, avroSchema, encoding).get("byKey"); + assertThat(byKey).hasSize(1); + final GenericRecord value = byKey.values().iterator().next(); + assertThat(value.get("label").toString()).isEqualTo("hits"); + assertThat(value.get("count")).isEqualTo(11); + } + + @ParameterizedTest + @EnumSource(AvroEncoding.class) + void testEnumWithReorderedFields(AvroEncoding encoding) throws Exception { + final Schema avroSchema = + record( + "{'name':'color','type':{'type':'enum','name':'Color','symbols':['RED','GREEN']}}", + "{'name':'name','type':'string'}"); + final RowType rowType = + rowType(FIELD("name", STRING().notNull()), FIELD("color", STRING().notNull())); + + final byte[] serialized = + serializer(rowType, avroSchema, encoding, FieldMatching.NAME) + .serialize( + GenericRowData.of( + StringData.fromString("Alice"), + StringData.fromString("GREEN"))); + + final GenericRecord written = decode(serialized, avroSchema, encoding); + assertThat(written.get("name").toString()).isEqualTo("Alice"); + assertThat(written.get("color").toString()).isEqualTo("GREEN"); + + final RowData roundTripped = + deserializer(rowType, avroSchema, encoding, FieldMatching.NAME) + .deserialize(serialized); + assertThat(roundTripped.getString(0).toString()).isEqualTo("Alice"); + assertThat(roundTripped.getString(1).toString()).isEqualTo("GREEN"); + } + + @ParameterizedTest + @EnumSource(AvroEncoding.class) + void testUnmatchedAvroFieldFallsBackToItsDefault(AvroEncoding encoding) throws Exception { + final Schema avroSchema = + record( + "{'name':'a','type':'string'}", + "{'name':'version','type':'int','default':7}"); + final RowType rowType = rowType(FIELD("a", STRING().notNull())); + + final GenericRecord written = + decode( + serializer(rowType, avroSchema, encoding, FieldMatching.NAME) + .serialize(GenericRowData.of(StringData.fromString("only"))), + avroSchema, + encoding); + + assertThat(written.get("a").toString()).isEqualTo("only"); + assertThat(written.get("version")).isEqualTo(7); + } + + @ParameterizedTest + @EnumSource(AvroEncoding.class) + void testColumnAbsentFromTheAvroSchemaIsReadAsNull(AvroEncoding encoding) throws Exception { + final Schema avroSchema = record("{'name':'a','type':'string'}"); + final RowType rowType = rowType(FIELD("a", STRING()), FIELD("b", INT())); + + final GenericRecord record = new GenericData.Record(avroSchema); + record.put("a", "present"); + + final RowData roundTripped = + deserializer(rowType, avroSchema, encoding, FieldMatching.NAME) + .deserialize(encode(record, avroSchema, encoding)); + + assertThat(roundTripped.getString(0).toString()).isEqualTo("present"); + assertThat(roundTripped.isNullAt(1)).isTrue(); + } + + @ParameterizedTest + @EnumSource(AvroEncoding.class) + void testColumnAbsentFromTheAvroSchemaCannotBeWritten(AvroEncoding encoding) throws Exception { + final Schema avroSchema = record("{'name':'a','type':'string'}"); + final RowType rowType = + rowType(FIELD("a", STRING().notNull()), FIELD("ghost", INT().notNull())); + + final AvroRowDataSerializationSchema serializer = + serializer(rowType, avroSchema, encoding, FieldMatching.NAME); + + assertThatThrownBy( + () -> + serializer.serialize( + GenericRowData.of(StringData.fromString("a"), 1))) + .isInstanceOf(RuntimeException.class) + .hasStackTraceContaining("Column 'ghost' cannot be written"); + } + + @ParameterizedTest + @EnumSource(AvroEncoding.class) + void testConvertersSurviveJavaSerialization(AvroEncoding encoding) throws Exception { + // The resolved pairing is transient, so a converter has to be able to rebuild it after + // being shipped to a task. + final Schema avroSchema = + record("{'name':'b','type':'int'}", "{'name':'a','type':'string'}"); + final RowType rowType = + rowType(FIELD("a", STRING().notNull()), FIELD("b", INT().notNull())); + + final AvroRowDataSerializationSchema serializer = + roundTripThroughJavaSerialization( + serializer(rowType, avroSchema, encoding, FieldMatching.NAME)); + final AvroRowDataDeserializationSchema deserializer = + roundTripThroughJavaSerialization( + deserializer(rowType, avroSchema, encoding, FieldMatching.NAME)); + serializer.open(null); + deserializer.open(null); + + final RowData roundTripped = + deserializer.deserialize( + serializer.serialize(GenericRowData.of(StringData.fromString("x"), 1))); + + assertThat(roundTripped.getString(0).toString()).isEqualTo("x"); + assertThat(roundTripped.getInt(1)).isEqualTo(1); + } + + // ------------------------------------------------------------------------ + // Utilities + // ------------------------------------------------------------------------ + + private static AvroRowDataSerializationSchema serializer( + RowType rowType, Schema avroSchema, AvroEncoding encoding, FieldMatching fieldMatching) + throws Exception { + final AvroRowDataSerializationSchema serializationSchema = + new AvroRowDataSerializationSchema( + rowType, + AvroSerializationSchema.forGeneric(avroSchema, encoding), + RowDataToAvroConverters.createConverter(rowType, true, fieldMatching)); + serializationSchema.open(null); + return serializationSchema; + } + + private static AvroRowDataDeserializationSchema deserializer( + RowType rowType, Schema avroSchema, AvroEncoding encoding, FieldMatching fieldMatching) + throws Exception { + final AvroRowDataDeserializationSchema deserializationSchema = + new AvroRowDataDeserializationSchema( + AvroDeserializationSchema.forGeneric(avroSchema, encoding), + AvroToRowDataConverters.createRowConverter( + avroSchema, rowType, fieldMatching), + InternalTypeInfo.of(rowType)); + deserializationSchema.open(null); + return deserializationSchema; + } + + private static T roundTripThroughJavaSerialization(T object) throws Exception { + return InstantiationUtil.deserializeObject( + InstantiationUtil.serializeObject(object), + Thread.currentThread().getContextClassLoader()); + } + + private static byte[] encode(GenericRecord record, Schema schema, AvroEncoding encoding) + throws Exception { + final ByteArrayOutputStream out = new ByteArrayOutputStream(); + final Encoder encoder = createEncoder(encoding, schema, out); + new GenericDatumWriter(schema).write(record, encoder); + encoder.flush(); + return out.toByteArray(); + } + + private static GenericRecord decode(byte[] bytes, Schema schema, AvroEncoding encoding) + throws Exception { + final AvroDeserializationSchema deserializationSchema = + AvroDeserializationSchema.forGeneric(schema, encoding); + deserializationSchema.open(null); + return deserializationSchema.deserialize(bytes); + } + + private static RowType rowType(org.apache.flink.table.api.DataTypes.Field... fields) { + return (RowType) ROW(fields).notNull().getLogicalType(); + } + + /** Builds a record schema from the given field declarations, using {@code '} for {@code "}. */ + private static Schema record(String... fields) { + return new Schema.Parser() + .parse( + ("{'type':'record','name':'TestRecord','fields':[" + + String.join(",", fields) + + "]}") + .replace('\'', '"')); + } +}