kumarpritam863 opened a new pull request, #28845:
URL: https://github.com/apache/flink/pull/28845
## 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.
- [FLINK-40263] Serializing a `CHAR`/`VARCHAR` column into an Avro
`ENUM` now validates the value. A declared symbol converts as before; an
unknown symbol falls back to the enum's declared `default` if it has one;
otherwise the conversion
fails with a message naming the value, the enum and the allowed symbols.
Previously the invalid symbol survived and failed later inside
`GenericDatumWriter` as `Not an enum`, naming none of them.
- [FLINK-40264] `AvroToRowDataConverters` now propagates
`legacyTimestampMapping` into nested row converters, so a nested
`TIMESTAMP_LTZ` column no longer fails with `Unsupported type: TIMESTAMP_LTZ`
when the non-legacy mapping is requested.
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.
- Extended `AvroRowDataSchemaProvidedSerDeSchemaTest` for the enum
default fallback and the invalid-symbol error message.
- Extended `AvroRowDataDeSerializationSchemaTest` with
`testTimestampTypeNewMappingInNestedRow`, which fails with `Unsupported type:
TIMESTAMP_LTZ(3) NOT NULL` without the `legacyTimestampMapping` propagation fix.
- Ran `mvn verify` for `flink-formats/flink-avro` and
`flink-formats/flink-avro-confluent-registry`: 406 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
