[FLINK-40262][format/avro] Support matching Avro record fields by name

[kumarpritam863]

What is the purpose of the change

RowDataToAvroConverters and AvroToRowDataConverters pair a Flink RowType field with an Avro record field by ordinal position. That is correct whenever the Avro schema is derived from the row type by AvroSchemaConverter, because
both sides then agree on field order by construction. It is not correct when the schema is supplied independently of the table:

- a schema registry subject,
- the `avro-confluent.schema` option,
- the `AvroRowDataSerializationSchema(rowType, nestedSchema, converter)` and `AvroRowDataDeserializationSchema(nestedSchema, converter, typeInfo)` constructors, which downstream connectors use directly.

There, column i and Avro field i need not be the same field, so values are written into and read from the wrong one. Where the types at a position differ the job fails with a ClassCastException; where they happen to agree, the data is
silently corrupted. A table a STRING, b STRING against a schema that declares b before a swaps the two values with no error at all.

This change lets the converters pair fields by name instead, behind an explicit opt-in that leaves the existing positional behaviour as the default.

Brief change log

- Added `FieldMatching`, a strategy with `INDEX` (the existing positional behaviour, and the default) and `NAME`.
- Threaded it through `RowDataToAvroConverters#createConverter` and `AvroToRowDataConverters#createRowConverter` as new overloads. Every pre-existing overload is kept and delegates with `INDEX`, so the change is binary compatible.
- Added `AvroFieldMatcher`, which resolves the pairing: exact name first, then Avro field aliases, then a case-insensitive comparison. Each stage runs to completion before the next, so a fuzzy match can never claim an Avro field that another

column matches exactly.
- AvroFieldMatcher rejects anything ambiguous or lossy: a column matching several Avro fields, two columns matching the same one, a column with no counterpart when writing, a NOT NULL column with no counterpart when reading, or an
unwritten Avro field that is neither nullable nor defaulted — which Avro would otherwise surface as a bare NullPointerException.
- Unwritten Avro fields that declare a default are written with that default; Avro fields that nothing reads are ignored, which is ordinary projection.
- Resolution runs once per (row type, schema) pair and is memoized, so NAME costs one array lookup per field per record: no hashing, no lowercasing and no allocation on the hot path. The memo is transient and rebuilt after the converter
is shipped to a task.

No SQL-facing option is added here. The converters are @Internal, so this change makes name matching reachable from Java only. Exposing it as an avro-confluent.field-matching option also requires relaxing the order-sensitive schema check
in RegistryAvroFormatFactory#getAvroSchema, which compares the converted DataType with equals() and therefore rejects a reordered schema during planning with "Schema provided for 'avro-confluent' format does not match the table
schema". That part is written and tested, but held back so the two can be reviewed independently — happy to add it to this PR if reviewers would rather have it in one go. The plain avro format needs no option, since it always derives its
schema from the table.

Verifying this change

This change added tests and can be verified as follows:

- Added `AvroFieldMatcherTest`: 15 unit tests over the resolution rules — reordering, case-insensitive matching, alias matching, exact-beats-fuzzy precedence, every rejection listed above, default materialization, and that a resolved plan is

bound to the schema it was resolved for.
- Added AvroRowDataFieldMatchingTest: 12 tests over both Avro encodings, end to end through AvroRowDataSerializationSchema and AvroRowDataDeserializationSchema — reordered top-level fields (asserted byte for byte against what a
GenericDatumWriter produces for the same schema), reordering inside nested rows, array elements and map values, case-insensitive and alias matching, enum combined with reordering, an unmatched Avro field falling back to its default, a column
absent from the schema read as NULL, a column absent from the schema refused on write, and a Java-serialization round trip of both converters, which exercises rebuilding the transient memo in a task.
- Added AvroRowDataFieldMatchingTest#testIndexMatchingIgnoresFieldNames, which pins down the behaviour NAME exists to avoid: with INDEX, a reordered schema of same-typed fields silently swaps the values.
- Ran mvn clean verify for flink-formats/flink-avro and flink-formats/flink-avro-confluent-registry: 403 and 27 tests respectively, no failures, 0 checkstyle violations, spotless and ArchUnit clean, and japicmp reports no
incompatibility.

Does this pull request potentially affect one of the following parts:

- Dependencies (does it add or upgrade a dependency): **no**
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: **yes** — `FieldMatching` is new and `@PublicEvolving`. No existing signature or default value is changed, and japicmp reports no incompatibility.
- The serializers: **yes** — `RowDataToAvroConverters` and `AvroToRowDataConverters`, but only behind `FieldMatching.NAME`. `INDEX` remains the default and its code path is a separate class, left untouched.
- The runtime per-record code paths (performance sensitive): **yes** — `NAME` adds one array lookup per field per record, with resolution memoized outside the record loop. `INDEX` is unaffected.
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: **no**
- The S3 file system connector: **no**

Documentation

- Does this pull request introduce a new feature? **yes**
- If yes, how is the feature documented? **JavaDocs** — nothing user-facing is added, since the strategy is only selectable from Java. User documentation lands with the `avro-confluent.field-matching` option in the follow-up.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Code (Anthropic Claude Opus 5)

---

[flinkbot]

CI report:

• a81e3dc Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build