hudi-agent commented on code in PR #19872:
URL: https://github.com/apache/hudi/pull/19872#discussion_r3998541521
##########
hudi-common/src/main/java/org/apache/hudi/common/model/BaseAvroPayload.java:
##########
@@ -126,24 +169,189 @@ protected boolean isEmptyRecord() {
}
protected Option<IndexedRecord> getRecord(Schema schema) throws IOException {
- if (record != null) {
- if (record.getSchema() == schema) {
- return Option.of(record);
- }
- // if the schema does not match, we need to deserialize with the proper
schema to match legacy behavior
- recordBytes = getRecordBytes();
+ if (record != null && record.getSchema() == schema) {
+ return Option.of(record);
}
- if (recordBytes == null || recordBytes.length == 0) {
+ byte[] bytes = getRecordBytes();
+ if (bytes.length == 0) {
return Option.empty();
}
- record = SerializableIndexedRecord.fromAvroBytes(schema, recordBytes);
+ if (writerSchema == null) {
+ // Legacy bytes have no schema. Preserve positional decoding until a
supplied schema
+ // consumes the complete record; a prefix projection must not become the
writer schema.
+ BinaryDecoder decoder = HoodieAvroUtils.getBinaryDecoder(bytes, 0,
bytes.length);
+ record = new GenericDatumReader<GenericRecord>(schema).read(null,
decoder);
+ if (decoder.isEnd()) {
+ writerSchema = schema;
+ }
+ return Option.of(record);
+ }
+ Map<String, String> renames = new HashMap<>();
+ collectRenames(writerSchema, schema, "", renames, new HashSet<>());
Review Comment:
🤖 This runs on every `getRecord` miss, which in `HoodieWriteMergeHandle` is
per incoming record (e.g. `isDelete` with the meta-field reader schema, then
the merge call) — each time walking every reader field, allocating a `Pair` and
a prefix string per field. Since `writerSchema` is normally one shared instance
per partition, have you considered memoizing the rename map keyed on the
identity pair `(writerSchema, schema)` (like Avro's own resolver cache) so the
walk happens once per schema pair rather than once per record?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-common/src/main/java/org/apache/hudi/common/model/BaseAvroPayload.java:
##########
@@ -126,24 +169,189 @@ protected boolean isEmptyRecord() {
}
protected Option<IndexedRecord> getRecord(Schema schema) throws IOException {
- if (record != null) {
- if (record.getSchema() == schema) {
- return Option.of(record);
- }
- // if the schema does not match, we need to deserialize with the proper
schema to match legacy behavior
- recordBytes = getRecordBytes();
+ if (record != null && record.getSchema() == schema) {
+ return Option.of(record);
}
- if (recordBytes == null || recordBytes.length == 0) {
+ byte[] bytes = getRecordBytes();
+ if (bytes.length == 0) {
return Option.empty();
}
- record = SerializableIndexedRecord.fromAvroBytes(schema, recordBytes);
+ if (writerSchema == null) {
+ // Legacy bytes have no schema. Preserve positional decoding until a
supplied schema
+ // consumes the complete record; a prefix projection must not become the
writer schema.
+ BinaryDecoder decoder = HoodieAvroUtils.getBinaryDecoder(bytes, 0,
bytes.length);
+ record = new GenericDatumReader<GenericRecord>(schema).read(null,
decoder);
+ if (decoder.isEnd()) {
+ writerSchema = schema;
+ }
+ return Option.of(record);
+ }
+ Map<String, String> renames = new HashMap<>();
+ collectRenames(writerSchema, schema, "", renames, new HashSet<>());
+ if (renames.isEmpty()) {
+ record = HoodieAvroUtils.bytesToAvro(bytes, writerSchema, schema);
+ } else {
+ GenericRecord original = HoodieAvroUtils.bytesToAvro(bytes,
writerSchema);
+ record = HoodieAvroUtils.rewriteRecordWithNewSchema(original, schema,
renames);
+ }
return Option.of(record);
}
+ private static void collectRenames(Schema writer, Schema reader, String
prefix,
+ Map<String, String> renames,
Set<Pair<Schema, Schema>> visiting) {
+ if (writer.equals(reader)) {
+ return;
+ }
+ Pair<Schema, Schema> pair = Pair.of(writer, reader);
+ if (!visiting.add(pair)) {
+ return;
+ }
+ try {
+ if (writer.getType() == Schema.Type.UNION || reader.getType() ==
Schema.Type.UNION) {
+ collectUnionRenames(writer, reader, prefix, renames, visiting);
+ } else if (writer.getType() == Schema.Type.RECORD && reader.getType() ==
Schema.Type.RECORD) {
+ collectRecordRenames(writer, reader, prefix, renames, visiting);
+ } else if (writer.getType() == Schema.Type.ARRAY && reader.getType() ==
Schema.Type.ARRAY) {
+ collectRenames(writer.getElementType(), reader.getElementType(),
prefix + "element.", renames, visiting);
+ } else if (writer.getType() == Schema.Type.MAP && reader.getType() ==
Schema.Type.MAP) {
+ collectRenames(writer.getValueType(), reader.getValueType(), prefix +
"value.", renames, visiting);
+ }
+ } finally {
+ visiting.remove(pair);
+ }
+ }
+
+ private static void collectUnionRenames(Schema writer, Schema reader, String
prefix,
+ Map<String, String> renames,
Set<Pair<Schema, Schema>> visiting) {
+ for (Schema writerBranch : writer.getType() == Schema.Type.UNION ?
writer.getTypes() : Collections.singletonList(writer)) {
+ for (Schema readerBranch : reader.getType() == Schema.Type.UNION ?
reader.getTypes() : Collections.singletonList(reader)) {
+ if (writerBranch.getType() == readerBranch.getType() &&
isSameRecordBranch(writerBranch, readerBranch, writer, reader)) {
+ collectRenames(writerBranch, readerBranch, prefix, renames,
visiting);
+ }
+ }
+ }
+ }
+
+ private static boolean isSameRecordBranch(Schema writerBranch, Schema
readerBranch, Schema writer, Schema reader) {
+ return writerBranch.getType() != Schema.Type.RECORD
+ || writerBranch.getFullName().equals(readerBranch.getFullName())
+ || readerBranch.getAliases().contains(writerBranch.getFullName())
+ || (hasSingleRecordBranch(writer) && hasSingleRecordBranch(reader));
+ }
+
+ private static boolean hasSingleRecordBranch(Schema schema) {
+ return schema.getType() != Schema.Type.UNION
+ || schema.getTypes().stream().filter(branch -> branch.getType() ==
Schema.Type.RECORD).count() == 1;
+ }
+
+ private static void collectRecordRenames(Schema writer, Schema reader,
String prefix,
+ Map<String, String> renames,
Set<Pair<Schema, Schema>> visiting) {
+ for (Schema.Field field : reader.getFields()) {
+ Schema.Field source = writer.getField(field.name());
+ if (source == null) {
+ for (String alias : field.aliases()) {
+ if (writer.getField(alias) != null) {
+ source = writer.getField(alias);
+ break;
+ }
+ }
+ }
+ // Keep the historical same-position rename only for compatible,
unclaimed fields.
+ // An incompatible drop/add must use its default; aliases must not
duplicate source values.
+ if (source == null && field.defaultVal() != null
Review Comment:
🤖 nit: this positional-fallback block (defaulted field + same field count +
unclaimed + compatible) is the trickiest rule in the method; could you pull it
into a small helper like `findPositionalFallback(writer, reader, field)`
returning `Option<Schema.Field>` so the name/alias/position resolution order
reads top-down?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-common/src/main/java/org/apache/hudi/common/model/BaseAvroPayload.java:
##########
@@ -126,24 +169,189 @@ protected boolean isEmptyRecord() {
}
protected Option<IndexedRecord> getRecord(Schema schema) throws IOException {
- if (record != null) {
- if (record.getSchema() == schema) {
- return Option.of(record);
- }
- // if the schema does not match, we need to deserialize with the proper
schema to match legacy behavior
- recordBytes = getRecordBytes();
+ if (record != null && record.getSchema() == schema) {
+ return Option.of(record);
}
- if (recordBytes == null || recordBytes.length == 0) {
+ byte[] bytes = getRecordBytes();
+ if (bytes.length == 0) {
return Option.empty();
}
- record = SerializableIndexedRecord.fromAvroBytes(schema, recordBytes);
+ if (writerSchema == null) {
Review Comment:
🤖 nit: it might be worth extracting the `writerSchema == null` legacy
positional-decode branch into e.g. `decodeLegacyBytes(schema, bytes)` so
`getRecord` reads as: cached fast path → legacy decode → name-based decode with
renames.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
