hudi-agent commented on code in PR #19872:
URL: https://github.com/apache/hudi/pull/19872#discussion_r3963976509
##########
hudi-common/src/main/java/org/apache/hudi/common/model/BaseAvroPayload.java:
##########
@@ -126,24 +154,129 @@ 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 serialized payloads did not include a writer schema.
+ writerSchema = schema;
+ }
+ 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) {
+ for (Schema w : writer.getType() == Schema.Type.UNION ?
writer.getTypes() : java.util.Collections.singletonList(writer)) {
+ for (Schema r : reader.getType() == Schema.Type.UNION ?
reader.getTypes() : java.util.Collections.singletonList(reader)) {
+ boolean sameRecordBranch = w.getType() != Schema.Type.RECORD ||
r.getType() != Schema.Type.RECORD
+ || w.getFullName().equals(r.getFullName()) ||
r.getAliases().contains(w.getFullName())
+ || (writer.getType() != Schema.Type.UNION ||
writer.getTypes().stream().filter(s -> s.getType() ==
Schema.Type.RECORD).count() == 1)
+ && (reader.getType() != Schema.Type.UNION ||
reader.getTypes().stream().filter(s -> s.getType() ==
Schema.Type.RECORD).count() == 1);
+ if (w.getType() == r.getType() && sameRecordBranch) {
+ collectRenames(w, r, prefix, renames, visiting);
+ }
+ }
+ }
+ } else if (writer.getType() == Schema.Type.RECORD && reader.getType() ==
Schema.Type.RECORD) {
+ 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;
+ }
+ }
+ }
+ // Preserve the historical rename contract only for a defaulted
field replacing a removed
+ // field at the same position. Added fields and named projections
must never shift values.
+ if (source == null && field.defaultVal() != null
+ && writer.getFields().size() == reader.getFields().size()) {
Review Comment:
🤖 This fallback fires for any drop+add at the same position with equal field
counts, not just renames — e.g. writer `{a, b: string}` → reader `{a, c: [null,
long] default null}` maps `c ← b`, and `rewritePrimaryType` then throws
`HoodieAvroSchemaException` where plain Avro resolution would have given `c =
null`. Should the fallback additionally require the candidate's schema to be
compatible with (or equal to) the reader field's schema before treating it as a
rename?
<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 +154,129 @@ 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 serialized payloads did not include a writer schema.
+ writerSchema = schema;
Review Comment:
🤖 For legacy-format payloads (Spark shuffle, old Java streams) the first
requested schema becomes the permanent writer schema. If that first request is
a positionally-compatible prefix (e.g. missing a trailing nullable column), a
later request with the wider schema now resolves by name from the narrow schema
and fills the trailing column with its default, whereas the old positional
decode would have read the real value from the bytes. Is there any path where
the first `getRecord` after a shuffle isn't the full record schema?
<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 +154,129 @@ 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 serialized payloads did not include a writer schema.
+ writerSchema = schema;
+ }
+ 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) {
+ for (Schema w : writer.getType() == Schema.Type.UNION ?
writer.getTypes() : java.util.Collections.singletonList(writer)) {
+ for (Schema r : reader.getType() == Schema.Type.UNION ?
reader.getTypes() : java.util.Collections.singletonList(reader)) {
+ boolean sameRecordBranch = w.getType() != Schema.Type.RECORD ||
r.getType() != Schema.Type.RECORD
+ || w.getFullName().equals(r.getFullName()) ||
r.getAliases().contains(w.getFullName())
Review Comment:
🤖 nit: `java.util.Collections.singletonList` is used fully-qualified inline
here — worth adding a proper import (or use `Collections.singletonList`) for
consistency with the rest of the file's style.
<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 +154,129 @@ 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 serialized payloads did not include a writer schema.
+ writerSchema = schema;
+ }
+ 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) {
+ for (Schema w : writer.getType() == Schema.Type.UNION ?
writer.getTypes() : java.util.Collections.singletonList(writer)) {
+ for (Schema r : reader.getType() == Schema.Type.UNION ?
reader.getTypes() : java.util.Collections.singletonList(reader)) {
+ boolean sameRecordBranch = w.getType() != Schema.Type.RECORD ||
r.getType() != Schema.Type.RECORD
+ || w.getFullName().equals(r.getFullName()) ||
r.getAliases().contains(w.getFullName())
+ || (writer.getType() != Schema.Type.UNION ||
writer.getTypes().stream().filter(s -> s.getType() ==
Schema.Type.RECORD).count() == 1)
+ && (reader.getType() != Schema.Type.UNION ||
reader.getTypes().stream().filter(s -> s.getType() ==
Schema.Type.RECORD).count() == 1);
+ if (w.getType() == r.getType() && sameRecordBranch) {
+ collectRenames(w, r, prefix, renames, visiting);
+ }
+ }
+ }
+ } else if (writer.getType() == Schema.Type.RECORD && reader.getType() ==
Schema.Type.RECORD) {
+ 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;
+ }
+ }
+ }
+ // Preserve the historical rename contract only for a defaulted
field replacing a removed
+ // field at the same position. Added fields and named projections
must never shift values.
+ if (source == null && field.defaultVal() != null
+ && writer.getFields().size() == reader.getFields().size()) {
+ Schema.Field candidate = writer.getFields().get(field.pos());
+ if (reader.getField(candidate.name()) == null) {
+ source = candidate;
+ }
+ }
+ if (source == null) {
+ if (field.defaultVal() == null) {
+ throw new AvroTypeException("Field '" + prefix + field.name() +
"' has no writer field or default");
+ }
+ continue;
+ }
+ if (!source.name().equals(field.name())) {
+ renames.put(prefix + field.name(), source.name());
+ }
+ collectRenames(source.schema(), field.schema(), prefix +
field.name() + ".", 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 void writeObject(ObjectOutputStream output) throws IOException {
+ getRecordBytes();
+ output.defaultWriteObject();
+ output.writeObject(writerSchema == null ? null :
SCHEMA_STRINGS.get(writerSchema));
+ }
+
+ private void readObject(ObjectInputStream input) throws IOException,
ClassNotFoundException {
+ input.defaultReadObject();
+ try {
+ String schemaJson = (String) input.readObject();
+ writerSchema = schemaJson == null ? null :
PARSED_SCHEMAS.get(schemaJson);
+ } catch (OptionalDataException e) {
+ if (!e.eof) {
+ throw e;
+ }
+ // Java-serialized payloads from before schema retention contain only
the default fields.
+ }
+ }
+
+ /**
+ * Selects the legacy payload format for a transport that supplies the
writer schema separately.
+ * Other Kryo instances retain a self-contained writer schema by default.
The setting survives
+ * graph resets and does not affect Java serialization or other Kryo
instances.
+ */
+ public static void useLegacyKryoFormat(Kryo kryo) {
+ kryo.getContext().put(KRYO_WRITE_LEGACY_FORMAT, Boolean.TRUE);
+ }
+
@Override
public void write(Kryo kryo, Output output) {
byte[] bytes = getRecordBytes();
- output.writeInt(bytes.length);
+ if (Boolean.TRUE.equals(kryo.getContext().get(KRYO_WRITE_LEGACY_FORMAT))) {
+ output.writeInt(bytes.length);
+ } else {
+ // Negative lengths distinguish schema-bearing payloads from the legacy
non-negative format.
+ output.writeInt(-bytes.length - 1);
+ output.writeString(writerSchema == null ? null :
SCHEMA_STRINGS.get(writerSchema));
Review Comment:
🤖 This new default format also applies to `SerializationUtils`' Kryo (used
by `DefaultSerializer` for `ExternalSpillableMap` spills in
`HoodieMergedLogRecordScanner` and `HoodieWriteMergeHandle`), and to Spark
shuffles whenever `spark.kryo.registrator` isn't set to
`HoodieSparkKryoRegistrar` (only the CLI and `UtilHelpers` wire it). In both
cases the full schema JSON is written per record — for a wide table that can be
10x+ the record bytes. Since those readers always supply the schema, could we
also call `useLegacyKryoFormat` in
`SerializationUtils.KryoInstantiator`/`HoodieCommonKryoRegistrar`, or dedupe
the schema per stream?
<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 +154,129 @@ 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 serialized payloads did not include a writer schema.
+ writerSchema = schema;
+ }
+ Map<String, String> renames = new HashMap<>();
Review Comment:
🤖 nit: `collectRenames` handles unions, records, arrays, and maps, plus
cycle-guarding and alias/positional-rename fallback logic all in one ~65 line
recursive method — might be worth splitting the union-handling and
record-handling branches into their own private methods for readability.
<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 +154,129 @@ 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 serialized payloads did not include a writer schema.
+ writerSchema = schema;
+ }
+ 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) {
+ for (Schema w : writer.getType() == Schema.Type.UNION ?
writer.getTypes() : java.util.Collections.singletonList(writer)) {
+ for (Schema r : reader.getType() == Schema.Type.UNION ?
reader.getTypes() : java.util.Collections.singletonList(reader)) {
+ boolean sameRecordBranch = w.getType() != Schema.Type.RECORD ||
r.getType() != Schema.Type.RECORD
+ || w.getFullName().equals(r.getFullName()) ||
r.getAliases().contains(w.getFullName())
+ || (writer.getType() != Schema.Type.UNION ||
writer.getTypes().stream().filter(s -> s.getType() ==
Schema.Type.RECORD).count() == 1)
+ && (reader.getType() != Schema.Type.UNION ||
reader.getTypes().stream().filter(s -> s.getType() ==
Schema.Type.RECORD).count() == 1);
Review Comment:
🤖 nit: this union-branch matching condition is pretty dense (nested
ternaries + stream counts inline). Consider pulling it into a small named
helper like `isSameRecordBranch(w, r, writer, reader)` so the intent (match
same-named/aliased record branches, or the sole record branch when unambiguous)
is easier to follow.
<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]
