voonhous commented on code in PR #19583:
URL: https://github.com/apache/hudi/pull/19583#discussion_r3765203932
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieMergeOnReadRDDV2.scala:
##########
@@ -148,12 +148,19 @@ class HoodieMergeOnReadRDDV2(@transient sc: SparkContext,
}
}
+ // The plain skip-merging reader cannot read a SHREDDED variant base file:
it requests native
+ // VariantType, which clips the shredded group to {metadata, value} and
reads value=null (the
+ // #19556 defect family). Splits with variant columns take the file-group
reader below, whose
+ // reader context requests the full-variant projection shape instead
(#19578).
+ private val requiredSchemaHasVariant: Boolean =
+ requiredSchema.structTypeSchema.fields.exists(f =>
sparkAdapter.isVariantType(f.dataType))
Review Comment:
Addressed. The routing now keys off
`sparkAdapter.buildFullVariantReadSchema(requiredSchema.structTypeSchema).isDefined`,
which subsumes the presence check too -- it returns `Some` only when there is
a top-level variant column *and* the adapter can build the shape, so 4.0 and
3.x keep the skip-merging reader.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieMergeOnReadRDDV2.scala:
##########
@@ -148,12 +148,19 @@ class HoodieMergeOnReadRDDV2(@transient sc: SparkContext,
}
}
+ // The plain skip-merging reader cannot read a SHREDDED variant base file:
it requests native
+ // VariantType, which clips the shredded group to {metadata, value} and
reads value=null (the
+ // #19556 defect family). Splits with variant columns take the file-group
reader below, whose
+ // reader context requests the full-variant projection shape instead
(#19578).
+ private val requiredSchemaHasVariant: Boolean =
+ requiredSchema.structTypeSchema.fields.exists(f =>
sparkAdapter.isVariantType(f.dataType))
+
override def compute(split: Partition, context: TaskContext):
Iterator[InternalRow] = {
val partition = split.asInstanceOf[HoodieMergeOnReadPartition]
val bytesReadCallback =
HoodieSparkInputMetricsUtils.getFSBytesReadOnThreadCallback()
val iter: Iterator[InternalRow] = partition.split match {
- case dataFileOnlySplit if dataFileOnlySplit.logFiles.isEmpty =>
+ case dataFileOnlySplit if dataFileOnlySplit.logFiles.isEmpty &&
!requiredSchemaHasVariant =>
Review Comment:
Addressed -- widening it was not intended. Confirmed the branch never
appends: `SparkFileFormatInternalRowReaderContext` builds the `PartitionedFile`
with `InternalRow.empty` and passes an empty partition schema, and the
builder's `withPartitionPath` only feeds the bootstrap merge.
The fast path is now kept whenever the split carries partition values parsed
off the path (`dataFile.exists(_.partitionValues.numFields > 0)`), which is
exactly the condition under which that reader appends them. So a
`drop.partition.columns` table keeps its current behaviour rather than trading
null variants for null partition columns.
The pre-existing gap on the merged branch is untouched -- the real fix there
(prune + re-append, plus a partition-path source for log-only slices, which
have no base `PartitionedFile` at all) is bigger than this PR and belongs in
its own issue.
--
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]
