voonhous commented on code in PR #19583:
URL: https://github.com/apache/hudi/pull/19583#discussion_r3765718482
##########
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:
Both halves addressed in 963eb6f5, and you were asking for the right fixture.
The carve-out itself is gone: it only existed because the file-group-reader
branch dropped partition values, and now that the branch appends them there is
no trade-off left, so the re-route is unconditional again.
The coverage you describe is what the fix is tested with.
`TestLegacyParquetReadPath` gained a MOR read over a `drop.partition.columns`
table through the directly-constructed legacy relation, with a partial upsert
so one file group merges base plus log while the others stay base-only -- the
merging branch and the fast path in one query. Verified red before the change
(the base-served p0 rows came back null) and green after. There is a second one
over a log-only slice; it passes either way, since HUDI-6926 makes a MOR upsert
ignore `drop.partition.columns` so the log records still carry the column, and
the test says so rather than implying it guards the null case.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/cdc/InternalRowToJsonStringConverter.scala:
##########
@@ -90,6 +92,11 @@ class InternalRowToJsonStringConverter(schema: StructType) {
structMap.toMap
case _ => value // fallback
}
+ case dt if SparkAdapterSupport.sparkAdapter.isVariantType(dt) =>
+ // VariantVal.toString renders the variant as JSON; embed it as a
real JSON node so
+ // the image carries the variant's structure. Falling through to the
default would
+ // serialize the VariantVal bean, i.e. its raw value/metadata bytes
as base64.
+ mapper.readTree(value.toString)
Review Comment:
Good catch on the rendering -- confirmed against the bytecode and by
rendering hand-built variants on 4.0.2 and 4.1.1: `Variant.toJsonImpl` guards
the double and float arms with `isFinite` and sends the non-finite one through
`appendQuoted`, so `NaN`/`Infinity` come out quoted and `readTree` never sees a
bare token. The comment and the test were both asserting something production
cannot produce; fixed in 6c439cfe.
The fallback does still earn its place, just for a different reason.
Jackson's default `StreamReadConstraints` cap field names at 50k chars, strings
at 20M and nesting at 1000 levels, and a variant can carry all three well
inside its own 128MB limit (`castToVariant` gates none of them, unlike
`parse_json`). I reproduced all three; each arrives as
`StreamConstraintsException`, which is a `JsonProcessingException`, so the
existing catch covers them. The comment now names those instead, and the test
is retargeted onto over-deep nesting -- it asserts the value as a JSON
*string*, so it only passes when the fallback actually fires.
On `toString` sitting outside the `try`: that one I kept, and documented
why. It throws `MALFORMED_VARIANT` (a `SparkRuntimeException`, disjoint from
`JsonProcessingException`, as you say) on corrupt bytes. A constraints failure
means "this valid variant is bigger than the parser's comfort zone", worth
degrading; `MALFORMED_VARIANT` with SQLSTATE 22023 means the bytes on disk are
corrupt, which an operator needs to see rather than have folded into an image
-- and if `toString` threw there is no rendering left to fall back to. The
honest alternatives would be `null`, which `Include.NON_ABSENT` then drops from
the image entirely so a corrupt variant is indistinguishable from a SQL NULL,
or an invented sentinel indistinguishable from real string data.
--
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]
