wombatu-kun commented on code in PR #19583:
URL: https://github.com/apache/hudi/pull/19583#discussion_r3765538280
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieMergeOnReadRDDV2.scala:
##########
@@ -148,12 +148,29 @@ 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). Such splits take the file-group reader below,
whose reader context
+ // requests the full-variant projection shape instead (#19578). Keyed off
the adapter building
+ // that shape rather than the mere presence of a variant column: it is None
below Spark 4.1,
+ // where the file-group reader would read the same nulls, so re-routing
there would cost the
+ // fast path for nothing.
+ private val shouldRerouteVariantSplit: Boolean =
+
sparkAdapter.buildFullVariantReadSchema(requiredSchema.structTypeSchema).isDefined
+
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 =>
+ // A split whose partition values were parsed off the partition path
keeps the fast path even
+ // when re-routing would apply: only that reader appends them
(drop.partition.columns,
+ // extract-from-path, bootstrap fast read), and the file-group reader
branch below has no
+ // equivalent, so re-routing would trade null variants for null
partition columns. Those
+ // splits stay on the pre-existing behaviour; the same gap on the merged
branch is older than
+ // this change and is tracked separately.
Review Comment:
Nothing tracks that gap - the only issues naming HoodieMergeOnReadRDDV2 are
#19465 and #19578, and neither mentions partition values. Either file one and
cite the number here, or reword to "left as a follow-up".
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieMergeOnReadRDDV2.scala:
##########
@@ -148,12 +148,29 @@ 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). Such splits take the file-group reader below,
whose reader context
+ // requests the full-variant projection shape instead (#19578). Keyed off
the adapter building
+ // that shape rather than the mere presence of a variant column: it is None
below Spark 4.1,
+ // where the file-group reader would read the same nulls, so re-routing
there would cost the
+ // fast path for nothing.
+ private val shouldRerouteVariantSplit: Boolean =
+
sparkAdapter.buildFullVariantReadSchema(requiredSchema.structTypeSchema).isDefined
+
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 =>
+ // A split whose partition values were parsed off the partition path
keeps the fast path even
+ // when re-routing would apply: only that reader appends them
(drop.partition.columns,
+ // extract-from-path, bootstrap fast read), and the file-group reader
branch below has no
+ // equivalent, so re-routing would trade null variants for null
partition columns. Those
+ // splits stay on the pre-existing behaviour; the same gap on the merged
branch is older than
+ // this change and is tracked separately.
+ case dataFileOnlySplit if dataFileOnlySplit.logFiles.isEmpty
+ && (!shouldRerouteVariantSplit ||
dataFileOnlySplit.dataFile.exists(_.partitionValues.numFields > 0)) =>
Review Comment:
Nothing exercises this carve-out - the new streaming test's table is
non-partitioned, so dropping the partitionValues condition would leave every
leg green. Worth a legacy-path read over a drop.partition.columns table, or is
that fixture out of scope here?
--
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]
