From a81e3dcc02f9d8293bbd4154387f8dbcbb73a43d Mon Sep 17 00:00:00 2001 From: Pritam Kumar Mishra Date: Thu, 30 Jul 2026 15:03:33 +0530 Subject: [PATCH] [FLINK-40262][format/avro] Support matching Avro record fields by name RowDataToAvroConverters and AvroToRowDataConverters pair a RowType field with an Avro record field by ordinal position. That is correct when the Avro schema is derived from the row type by AvroSchemaConverter, since both sides then agree on field order, but not when the schema is supplied independently - a schema registry subject, 'avro-confluent.schema', or the AvroRowData{,De}serialization Schema constructors that take a nested schema. There, column i and Avro field i need not be the same field, and the converters silently write and read the wrong one. Introduce an explicit strategy, FieldMatching, and thread it through both converters: - INDEX keeps today's behaviour and stays the default, byte for byte. - NAME pairs fields by name. Names are compared exactly first, then against Avro field aliases, and finally ignoring case, each stage running to completion before the next so a fuzzy match can never claim a field that some other column matches exactly. AvroFieldMatcher does the resolution and refuses anything that is ambiguous or that would quietly lose data: a column matching several Avro fields, two columns matching the same one, a column with no Avro counterpart when writing, a NOT NULL column with no counterpart when reading, or an Avro field that nothing writes to and that is neither nullable nor defaulted. Avro fields that no column maps to but that declare a default are written with that default; Avro fields no column reads are ignored, which is ordinary projection. Resolution is done once per (row type, schema) pair and memoized, so NAME costs an array lookup per field per record - no hashing, no lowercasing, no allocation on the hot path. The memo is transient and rebuilt after the converter is shipped to a task. This commit only makes the converters capable of it. Exposing the strategy to SQL as an 'avro-confluent.field-matching' option, and relaxing the order-sensitive schema check in RegistryAvroFormatFactory that rejects such a schema outright today, is left to a follow-up so the two can be reviewed independently. Restructuring createRowConverter into createRowConverterInternal necessarily threads legacyTimestampMapping into nested converters, which incidentally fixes FLINK-40264. That is submitted separately together with its regression test; whichever of the two lands second needs a trivial rebase. --- .../flink/formats/avro/AvroFieldMatcher.java | 348 +++++++++++++ .../formats/avro/AvroToRowDataConverters.java | 242 ++++++++- .../flink/formats/avro/FieldMatching.java | 69 +++ .../formats/avro/RowDataToAvroConverters.java | 160 ++++-- .../formats/avro/AvroFieldMatcherTest.java | 217 ++++++++ .../avro/AvroRowDataFieldMatchingTest.java | 465 ++++++++++++++++++ 6 files changed, 1453 insertions(+), 48 deletions(-) create mode 100644 flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroFieldMatcher.java create mode 100644 flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/FieldMatching.java create mode 100644 flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroFieldMatcherTest.java create mode 100644 flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroRowDataFieldMatchingTest.java 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('\'', '"')); + } +}