wombatu-kun commented on code in PR #18923:
URL: https://github.com/apache/hudi/pull/18923#discussion_r3859930170
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/SparkFileFormatInternalRowReaderContext.scala:
##########
@@ -96,29 +106,46 @@ class
SparkFileFormatInternalRowReaderContext(baseFileReader: SparkColumnarFileR
})
}
- // Aligns log-block records with the PushVariantIntoScan-projected variant
shape before
- // they reach the merger. Preserves merger metadata cols (_hoodie_record_key,
- // _tmp_metadata_row_index) which the merger reads by ordinal — projecting
down to the
- // bare required schema would drop them and the merger would read garbage
offsets.
- override def getLogBlockRecordProjection(
- dataBlockSchema: HoodieSchema): HOption[JFunction[InternalRow,
InternalRow]] = {
- val needsProjection = sparkRequiredSchema.exists(_.fields.exists(f =>
f.dataType match {
+ // True only when there is a Spark 4.1 PushVariantIntoScan projection to
apply AND the table is
+ // not using a custom (payload-based) merger. Payload-based tables
round-trip records through
+ // PayloadUpdateProcessor.convertToAvroRecord against a schema that still
types variant fields as
+ // VariantType, so a row already rewritten into the projected struct shape
would be mis-decoded.
+ // Single source of truth for both reader paths (parquet native projection +
avro rewrite).
+ private def shouldProjectVariants(): Boolean = {
+ val hasVariantProjection =
sparkRequiredSchema.exists(_.fields.exists(_.dataType match {
case st: StructType => sparkAdapter.isVariantProjectionStruct(st)
case _ => false
}))
- if (!needsProjection) {
- return HOption.empty[JFunction[InternalRow, InternalRow]]()
+ // getRecordMerger() is a Lombok getter over a field initialized to null
(not Option.empty());
+ // it stays null until setRecordMerger() runs during reader init, so the
null guard is required.
Review Comment:
This points at `setRecordMerger()`, but that Lombok setter has no production
callers - the field is assigned by `HoodieReaderContext.initRecordMerger`,
which `HoodieFileGroupReader` calls from its constructor. Naming
`initRecordMerger` instead would send the next reader to the real
initialization point.
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/SparkFileFormatInternalRowReaderContext.scala:
##########
@@ -24,13 +24,14 @@ import
org.apache.hudi.SparkFileFormatInternalRowReaderContext.{filterIsSafeForB
import org.apache.hudi.common.engine.HoodieReaderContext
import org.apache.hudi.common.fs.FSUtils
import org.apache.hudi.common.model.{HoodieFileFormat, HoodieRecord}
+import
org.apache.hudi.common.model.HoodieRecordMerger.PAYLOAD_BASED_MERGE_STRATEGY_UUID
import org.apache.hudi.common.schema.{HoodieSchema, HoodieSchemaUtils}
import org.apache.hudi.common.table.HoodieTableConfig
import
org.apache.hudi.common.table.read.buffer.PositionBasedFileGroupRecordBuffer.ROW_INDEX_TEMPORARY_COLUMN_NAME
import org.apache.hudi.common.util.HoodieVectorUtils
import org.apache.hudi.common.util.{Option => HOption}
Review Comment:
`HOption` is now unused - its only call sites were inside the deleted
`getLogBlockRecordProjection`, and the two remaining hudi `Option` references
in this file are fully qualified. Drop the alias from the import list;
scalastyle has no unused-import rule and the build passes no
`-Ywarn-unused-import`, so nothing else will flag it.
##########
hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieAvroDataBlock.java:
##########
@@ -170,7 +170,10 @@ protected <T> ClosableIterator<HoodieRecord<T>>
deserializeRecords(
protected <T> ClosableIterator<T> deserializeRecords(HoodieReaderContext<T>
readerContext, byte[] content) throws IOException {
checkState(this.readerSchema != null, "Reader's schema has to be
non-null");
RecordIterator iterator = RecordIterator.getInstance(this, content,
readerContext.enableLogicalTimestampFieldRepair());
- return new CloseableMappingIterator<>(iterator, data ->
readerContext.getRecordContext().convertAvroRecord(data));
+ ClosableIterator<T> records = new CloseableMappingIterator<>(iterator,
data -> readerContext.getRecordContext().convertAvroRecord(data));
+ // Align records with the engine's projected read schema (e.g. Spark 4.1
PushVariantIntoScan).
+ // No-op for engines/queries that don't need it. Parquet log blocks
project natively in the reader.
+ return readerContext.projectLogBlockRecords(records, this.readerSchema);
Review Comment:
On the current write table version `AppendHandleFactory` picks
`HoodieNativeLogAppendHandle`, which takes the log format from
`getBaseFileFormat()` and ignores `hoodie.logfile.data.block.format`, so
`withRecordType()`'s avro leg in `TestVariantDataType` still writes native
parquet logs and never reaches this hook with the gate open. Could a MOR case
pinned to a pre-native `hoodie.write.table.version` cover the avro leg, so a
regression in `projectLogBlockRecords` would not be silent?
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/SparkFileFormatInternalRowReaderContext.scala:
##########
@@ -96,29 +106,46 @@ class
SparkFileFormatInternalRowReaderContext(baseFileReader: SparkColumnarFileR
})
}
- // Aligns log-block records with the PushVariantIntoScan-projected variant
shape before
- // they reach the merger. Preserves merger metadata cols (_hoodie_record_key,
- // _tmp_metadata_row_index) which the merger reads by ordinal — projecting
down to the
- // bare required schema would drop them and the merger would read garbage
offsets.
- override def getLogBlockRecordProjection(
- dataBlockSchema: HoodieSchema): HOption[JFunction[InternalRow,
InternalRow]] = {
- val needsProjection = sparkRequiredSchema.exists(_.fields.exists(f =>
f.dataType match {
+ // True only when there is a Spark 4.1 PushVariantIntoScan projection to
apply AND the table is
+ // not using a custom (payload-based) merger. Payload-based tables
round-trip records through
+ // PayloadUpdateProcessor.convertToAvroRecord against a schema that still
types variant fields as
+ // VariantType, so a row already rewritten into the projected struct shape
would be mis-decoded.
+ // Single source of truth for both reader paths (parquet native projection +
avro rewrite).
+ private def shouldProjectVariants(): Boolean = {
+ val hasVariantProjection =
sparkRequiredSchema.exists(_.fields.exists(_.dataType match {
case st: StructType => sparkAdapter.isVariantProjectionStruct(st)
case _ => false
}))
- if (!needsProjection) {
- return HOption.empty[JFunction[InternalRow, InternalRow]]()
+ // getRecordMerger() is a Lombok getter over a field initialized to null
(not Option.empty());
+ // it stays null until setRecordMerger() runs during reader init, so the
null guard is required.
+ val merger = getRecordMerger()
+ val isPayloadBased = merger != null && merger.isPresent &&
merger.get.getMergingStrategy == PAYLOAD_BASED_MERGE_STRATEGY_UUID
+ hasVariantProjection && !isPayloadBased
Review Comment:
On a payload-based table this predicate keeps log rows as full variants, but
the `structType` overlay in `getFileRecordIterator` still projects base rows
unconditionally, so the two sides reach the merger in different shapes. Is
leaving the base overlay outside this gate intentional - follow-up rather than
a blocker if so?
--
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]
