voonhous commented on code in PR #18923:
URL: https://github.com/apache/hudi/pull/18923#discussion_r3870897561
##########
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:
Done, dropped the alias in e11af7e12adc.
##########
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:
Done, the comment now names HoodieReaderContext.initRecordMerger and the
HoodieFileGroupReader constructor that calls it (e11af7e12adc).
##########
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:
Not new to this PR: the overlay in getFileRecordIterator has been
unconditional since #18674, and the log-side predicate here is the one the
buffer used before (getPayloadClasses derives payloadClasses.isPresent() from
the PAYLOAD_BASED strategy). Gating the overlay too would not make the pair
coherent: appendPartitionAndProject projects reader output to the
post-PushVariantIntoScan shape, so base rows must leave the reader projected,
and log-only records on a payload table already come back VariantVal-shaped
from PayloadUpdateProcessor. Payload-based merging under PushVariantIntoScan
needs its own fix; filed #19768 and leaving it out of this PR.
##########
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:
Right, on table version 10 HoodieNativeLogAppendHandle takes the format from
getBaseFileFormat() and never reads hoodie.logfile.data.block.format, so that
leg was writing parquet logs. Added a MOR case pinned to
hoodie.write.table.version = 9 under the AVRO record type (e50c51b4d473). It
reads the log blocks back and asserts AVRO_DATA_BLOCK, so a silent fallback to
parquet fails instead of passing, then checks the base + avro-log merge and a
delete through cast(v as string).
--
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]
