hudi-agent commented on code in PR #19280:
URL: https://github.com/apache/hudi/pull/19280#discussion_r3624702730
##########
hudi-common/src/main/java/org/apache/hudi/common/model/debezium/PostgresDebeziumAvroPayload.java:
##########
@@ -96,22 +95,31 @@ public Option<IndexedRecord>
combineAndGetUpdateValue(IndexedRecord currentValue
Option<IndexedRecord> insertOrDeleteRecord =
super.combineAndGetUpdateValue(currentValue, schema);
if (insertOrDeleteRecord.isPresent()) {
- mergeToastedValuesIfPresent(insertOrDeleteRecord.get(), currentValue);
+ return Option.of(mergeToastedValuesIfPresent(insertOrDeleteRecord.get(),
currentValue));
}
return insertOrDeleteRecord;
}
- private void mergeToastedValuesIfPresent(IndexedRecord incomingRecord,
IndexedRecord currentRecord) {
- List<Schema.Field> fields = incomingRecord.getSchema().getFields();
-
- fields.forEach(field -> {
+ /**
+ * Fills any TOASTed sentinel columns in {@code incomingRecord} with the
value from
+ * {@code currentRecord}. Returns a NEW record when a fill occurs, so a
caller (e.g. the record
+ * merger) does not mistake the result for the unchanged incoming record and
discard the fill;
+ * returns {@code incomingRecord} unchanged when there is nothing to fill.
+ */
+ private IndexedRecord mergeToastedValuesIfPresent(IndexedRecord
incomingRecord, IndexedRecord currentRecord) {
+ GenericData.Record filled = null;
+ for (Schema.Field field : incomingRecord.getSchema().getFields()) {
// There are only four avro data types that have unconstrained sizes,
which are
// NON-NULLABLE STRING, NULLABLE STRING, NON-NULLABLE BYTES, NULLABLE
BYTES
if (((GenericRecord) incomingRecord).get(field.name()) != null
&& (containsStringToastedValues(incomingRecord, field) ||
containsBytesToastedValues(incomingRecord, field))) {
- ((GenericRecord) incomingRecord).put(field.name(),
((GenericData.Record) currentRecord).get(field.name()));
+ if (filled == null) {
+ filled = new GenericData.Record((GenericData.Record) incomingRecord,
false);
Review Comment:
🤖 I think this concern is well-founded. `BaseAvroPayload.getRecord()` does
`record = SerializableIndexedRecord.fromAvroBytes(schema, recordBytes)`
whenever the stored record's schema isn't reference-equal to the requested one
(the `record.getSchema() == schema` check is identity, not `.equals()`), so
`super.combineAndGetUpdateValue(...).get()` really can hand back a
`SerializableIndexedRecord`. Worth noting the same casts already exist on the
bytes path — `containsBytesToastedValues` casts to `(GenericData.Record)` and
line 112 casts `currentRecord` to `(GenericData.Record)` — so a
copy-constructor cast wouldn't be the only exposure. Building the copy from
`incomingRecord.getSchema()` and copying field-by-field through the
`GenericRecord`/`IndexedRecord` interface would sidestep all of these. Might be
worth a test that forces the schema-mismatch deserialization path to confirm.
##########
hudi-common/src/main/java/org/apache/hudi/common/model/debezium/PostgresDebeziumAvroPayload.java:
##########
@@ -96,22 +95,31 @@ public Option<IndexedRecord>
combineAndGetUpdateValue(IndexedRecord currentValue
Option<IndexedRecord> insertOrDeleteRecord =
super.combineAndGetUpdateValue(currentValue, schema);
if (insertOrDeleteRecord.isPresent()) {
- mergeToastedValuesIfPresent(insertOrDeleteRecord.get(), currentValue);
+ return Option.of(mergeToastedValuesIfPresent(insertOrDeleteRecord.get(),
currentValue));
}
return insertOrDeleteRecord;
}
- private void mergeToastedValuesIfPresent(IndexedRecord incomingRecord,
IndexedRecord currentRecord) {
- List<Schema.Field> fields = incomingRecord.getSchema().getFields();
-
- fields.forEach(field -> {
+ /**
+ * Fills any TOASTed sentinel columns in {@code incomingRecord} with the
value from
+ * {@code currentRecord}. Returns a NEW record when a fill occurs, so a
caller (e.g. the record
+ * merger) does not mistake the result for the unchanged incoming record and
discard the fill;
+ * returns {@code incomingRecord} unchanged when there is nothing to fill.
+ */
+ private IndexedRecord mergeToastedValuesIfPresent(IndexedRecord
incomingRecord, IndexedRecord currentRecord) {
Review Comment:
🤖 nit: the method was `void` before and mutated in place; now it returns a
new record, but `mergeToastedValuesIfPresent` still reads like a void
side-effect call. Could you rename it to something like
`withToastedValuesFilled` or `fillToastedValues` to make the return-value
contract obvious to a future reader at the call site?
<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]
