[
https://issues.apache.org/jira/browse/FLINK-40262?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18100684#comment-18100684
]
Dale Lane commented on FLINK-40262:
-----------------------------------
Thanks for the writeup. The scoping to converters-only makes sense. I've been
reading it alongside the code and I've ended up questioning the Problem
section, which I'd like to raise here, please.
Two things I'd like to ask for, and then my reasoning:
* Could the Problem section be updated to name what actually produces the
mismatch? (As far as I can see the registry subject doesn't, in either
direction.)
* Is the read half motivated by {{{}enum{}}}/{{{}fixed{}}}, or by the registry
subject? If it's the latter, I think it could come out (which would take the
eager-resolution path and the reader-schema navigation helpers with it, and
leave a considerably smaller change)
*I don't think the registry subject can produce this*
Of the three sources listed, the registry subject is the one I'm bumping on.
{*}Reading{*}: with no {{avro-confluent.schema}} set, the reader schema is
derived from the row type
([RegistryAvroFormatFactory.java#L103-L106|https://github.com/apache/flink/blob/6ce85191d154c3691e0d514be770dfa9cc5f0c61/flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/RegistryAvroFormatFactory.java#L103-L106]),
and
[RegistryAvroDeserializationSchema.java#L97-L103|https://github.com/apache/flink/blob/6ce85191d154c3691e0d514be770dfa9cc5f0c61/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/RegistryAvroDeserializationSchema.java#L97-L103]
sets both the writer schema and the expected schema on the datum reader. So
Avro resolves the registry's writer schema against the reader schema by name
before the converter sees anything, and the record it hands over carries the
_reader's_ schema and the reader's field order — meaning the {{record.get( i
)}} at
[AvroToRowDataConverters.java#L90|https://github.com/apache/flink/blob/6ce85191d154c3691e0d514be770dfa9cc5f0c61/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroToRowDataConverters.java#L90]
is reading the column it thinks it is. (writer {{[b, a]}} into reader {{[a,
b]}} will come back in reader order.)
Projection pushdown is the same point from another angle.
{{createRuntimeDecoder}} projects the row type before deriving the schema
([RegistryAvroFormatFactory.java#L101-L106|https://github.com/apache/flink/blob/6ce85191d154c3691e0d514be770dfa9cc5f0c61/flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/RegistryAvroFormatFactory.java#L101-L106]),
so under any pushdown the reader schema has fewer fields, at different
positions, than whatever the registry is holding. If the converters were
exposed to registry field order, projection pushdown would already be broken.
{*}Writing{*}:
[ConfluentSchemaRegistryCoder.java#L85|https://github.com/apache/flink/blob/6ce85191d154c3691e0d514be770dfa9cc5f0c61/flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/ConfluentSchemaRegistryCoder.java#L85]
unconditionally registers whatever schema Flink hands it, which (in the
absence of a {{avro-confluent.schema}} value) will be derived from the row
type. There's no path that fetches an existing schema and conforms to it;
{{register()}} and {{getById()}} are the only registry calls in the module. A
reordered schema would just registered as a new version. Field order isn't part
of Avro's resolution rules, so I'd expect a compatibility check to accept it.
That leaves the two constructors:
*
[AvroRowDataDeserializationSchema.java#L120-L127|https://github.com/apache/flink/blob/6ce85191d154c3691e0d514be770dfa9cc5f0c61/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowDataDeserializationSchema.java#L120-L127]
*
[AvroRowDataSerializationSchema.java#L96-L103|https://github.com/apache/flink/blob/6ce85191d154c3691e0d514be770dfa9cc5f0c61/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowDataSerializationSchema.java#L96-L103]
The caller supplies the nested schema and separately supplies a converter built
from a {{{}RowType{}}}, with nothing tying the two together. On the write side,
the supplied schema is what the converter is handed
([AvroRowDataSerializationSchema.java#L119|https://github.com/apache/flink/blob/6ce85191d154c3691e0d514be770dfa9cc5f0c61/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowDataSerializationSchema.java#L119]),
so where a caller does construct that pair the corruption is real, and I agree
there's something here.
But inside Flink every caller either derives the schema from the row type or
validates a supplied one first: {{{}RegistryAvroFormatFactory{}}}, both
Debezium schemas (via {{validateSchemaString}} in their constructors),
{{AvroFileFormatFactory}} in both directions, and the plain {{avro}} format has
no schema option at all.
You mention connectors using those constructors. If that's where the real case
comes from, naming one would make the ticket clearer than the registry subject
does.
The third source, {{{}avro-confluent.schema{}}}, I'm not treating as one that
goes away: the Scope section says relaxing [that
check|https://github.com/apache/flink/blob/6ce85191d154c3691e0d514be770dfa9cc5f0c61/flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/RegistryAvroFormatFactory.java#L255-L260]
is part of the follow-up.
*Writing and reading look like different cases*
*Writing* is where I can most see this mattering. There's no reconciliation
step on the way out, and since {{writeSchema}} always calls {{register()}}
there's no fallback if the registry won't take the schema. Against a subject
that's locked down, the job just fails. Someone pinned to a governed subject
has no way to emit against that schema's field order today, and filling an
unmatched Avro field from its declared default fits that well.
*Reading* I'm less sure about, because Avro resolution covers part of it
already.
[AvroSchemaConverter.java#L554-L556|https://github.com/apache/flink/blob/6ce85191d154c3691e0d514be770dfa9cc5f0c61/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/AvroSchemaConverter.java#L554-L556]
gives every nullable column a {{{}default: null{}}}, so a writer schema
missing that field reads as null, and a {{NOT NULL}} column with no counterpart
throws. This is the same pair of outcomes your {{AvroFieldMatcher}} implements.
([AvroFileFormatFactory.java#L128-L130|https://github.com/apache/flink/blob/6ce85191d154c3691e0d514be770dfa9cc5f0c61/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroFileFormatFactory.java#L128-L130]
already leans on that.)
Where I think reading does have a case is a supplied reader schema, for two
reasons:
* {{enum}} and {{{}fixed{}}}. The converters can already cope with both (the
string converter emits an {{EnumSymbol}} (FLINK-39053) and the decimal
converter accepts a {{GenericFixed)}} but the derived reader schema can't
express either, and DECIMAL maps to {{bytes}}
([AvroSchemaConverter.java#L530-L536|https://github.com/apache/flink/blob/6ce85191d154c3691e0d514be770dfa9cc5f0c61/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/AvroSchemaConverter.java#L530-L536]),
so resolution fails first: {{{}AvroTypeException: Found Color, expecting
string{}}}, and {{{}Found Dec, expecting bytes{}}}. A hand-written
{{avro-confluent.schema}} gets the enum case working today provided the field
order matches the table (which is what the tests in
[#27558|https://github.com/apache/flink/pull/27558] exercise) — so what NAME is
adding is support for enum-or-fixed _and_ an order the table doesn't share.
* Avro resolution is case-sensitive, and its alias handling matches reader
aliases against writer field names, not against Flink column names. So your
fuzzy stages do add something a derived reader schema can't get.
Both of those only apply where the reader schema is supplied, though, which
lands back at the same two sources rather than at the registry subject.
Apologies if I've misread any of this, I might've gone down a rabbit-hole
thinking this through!
> Support matching Avro record fields by name in the Avro RowData converters
> --------------------------------------------------------------------------
>
> Key: FLINK-40262
> URL: https://issues.apache.org/jira/browse/FLINK-40262
> Project: Flink
> Issue Type: Improvement
> Components: Formats (JSON, Avro, Parquet, ORC, SequenceFile)
> Reporter: Pritam Kumar
> Assignee: Pritam Kumar
> Priority: Major
> Labels: pull-request-available
>
> h2. Problem
> {{RowDataToAvroConverters}} and {{AvroToRowDataConverters}} pair a
> {{RowType}} field with an Avro record field by *ordinal position*:
> {code:java}
> for (int i = 0; i < length; ++i) {
> final Schema.Field schemaField = fields.get(i);
> record.put(i, fieldConverters[i].convert(
> schemaField.schema(), fieldGetters[i].getFieldOrNull(row)));
> }
> {code}
> * 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.
> In those cases column {{i}} and Avro field {{i}} need not be the same
> field. Writing puts each value into whichever field happens to sit at that
> position, and reading does the mirror image. Where the types at a position
> differ the job fails
> with a {{ClassCastException}}; where they happen to agree, the data is
> silently corrupted.
> h3. Example
> Table {{a STRING, b STRING}} against a registry schema that declares {{b}}
> before {{a}}. Both are strings, so nothing fails: {{a}}'s value is written
> into {{b}} and vice versa.
> h2. Proposal
> Add an explicit strategy and thread it through both converters:
> {code:java}
> public enum FieldMatching { INDEX, NAME }
> {code}
> * {{INDEX}} — today's behaviour, and the default. Unchanged.
> * {{NAME}} — pair fields by name. Names are compared exactly first, then
> against Avro field {{aliases}}, and finally ignoring case ({{Locale.ROOT}}).
> Each stage runs to completion before the next, so a fuzzy match can never
> claim an Avro field
> that another column matches exactly.
> Every existing overload is kept and delegates with {{INDEX}}, so the change
> is binary compatible (japicmp clean).
> h3. What NAME refuses
> Name matching should not paper over a schema that does not line up.
> Resolution fails, naming the offending field, when:
> * a column matches more than one Avro field;
> * two columns match the same Avro field;
> * a column has no Avro counterpart and we are writing — the column would be
> dropped;
> * a {{NOT NULL}} column has no Avro counterpart and we are reading — it
> could only be read as {{NULL}};
> * an Avro field that no column maps to is neither nullable nor declares a
> default — the record cannot be encoded at all, and Avro surfaces that as a
> bare {{NullPointerException}}.
> Tolerated:
> * an Avro field that no column maps to but that declares a default is
> written with that default;
> * an Avro field that no column reads is ignored, which is ordinary
> projection.
> h3. Cost
> 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, no allocation on the hot path. The memo is {{transient}} and
> rebuilt after the converter is
> shipped to a task.
> h2. Scope
> This issue covers the converters only, which makes the capability reachable
> from Java but not from SQL.
> Exposing the strategy as an {{avro-confluent.field-matching}} option is
> deliberately left to a follow-up so that the two can be reviewed
> independently. That part also has to relax 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"_ — so today such a schema never reaches the converters at all.
> Happy to fold that into this issue instead if reviewers would rather have it
> in one go.
> The plain {{avro}} format needs no option: it always derives its schema
> from the table, so field order agrees by construction.
> h2. Compatibility
> {{INDEX}} is the default everywhere, no existing signature or default value
> changes, and japicmp reports no incompatibility.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
