voonhous commented on code in PR #19458:
URL: https://github.com/apache/hudi/pull/19458#discussion_r3701559540
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestNestedSchemaPruningOptimization.scala:
##########
@@ -58,26 +64,34 @@ class TestNestedSchemaPruningOptimization extends
HoodieSparkSqlTestBase {
}
}
- test("Test nested schema pruning with DefaultHoodieRecordPayload") {
+ test("Test nested schema pruning with a projection-incompatible custom
payload") {
withTempDir { tmp =>
val tableName = generateTableName
val tablePath = s"${tmp.getCanonicalPath}/$tableName"
- // NOTE: On the file-group-reader based read path the payload class does
not affect nested
- // schema pruning, so the read schema is pruned the same way as
with the default payload
+ // NOTE: A payload class outside the well-known set puts the table in
CUSTOM merge mode, whose
+ // merger is not projection compatible, so the file group reader
merges on the full
+ // table schema internally
(FileGroupReaderSchemaHandler#generateRequiredSchema) and
+ // projects the merged rows back down to the pruned read schema
afterwards
createTableWithNestedStructSchema("mor", tableName, tablePath,
- Map(HoodieWriteConfig.WRITE_PAYLOAD_CLASS_NAME.key ->
"org.apache.hudi.common.model.DefaultHoodieRecordPayload"))
+ Map(HoodieWriteConfig.WRITE_PAYLOAD_CLASS_NAME.key ->
classOf[CustomPayloadForTesting].getName),
+ populateMetaFields = true)
+
+ // The update writes a log file, so the pruned reads below actually
merge through that gate
+ spark.sql(s"UPDATE $tableName SET ts = 123457 WHERE id = 1")
val selectDF = spark.sql(s"SELECT id, item.name FROM $tableName")
+ // Spark still prunes the scan schema; the full-schema requirement is
internal to the reader
val expectedSchema = StructType(Seq(
StructField("id", IntegerType, nullable = true),
StructField("item", StructType(Seq(StructField("name", StringType,
nullable = false))), nullable = true)
))
-
assertPrunedReadSchema(selectDF, tableName, expectedSchema)
checkAnswer(s"SELECT id, item.name FROM $tableName")(Seq(1, "a1"))
+ // The merged row keeps nested leaves that the pruned read schema dropped
+ checkAnswer(s"SELECT id, item.price, ts FROM $tableName")(Seq(1, 10,
123457))
Review Comment:
Checked: no, it would not fail -- `SchemaHandlerTestBase` (plus
`TestFileGroupReaderSchemaHandler#testSchemaForMandatoryFields`, which calls
`generateRequiredSchema` directly) are the only pins.
Without the branch, the required schema falls back to requested + mandatory
merge fields, and the mandatory set already contains `ts` (the ordering field)
-- the only column `DefaultHoodieRecordPayload#needUpdatingPersistedRecord`
reads across records. `CustomPayloadForTesting` adds nothing over
`DefaultHoodieRecordPayload` and returns the incoming record verbatim, and the
MOR `UPDATE` writes full-row log blocks, so both `checkAnswer`s pass either way.
So this suite covers the full-schema-merge-then-project round trip through
Spark rather than discriminating the branch; the PR description now says as
much. A true E2E discriminator would need a merger that reads a non-mandatory,
non-projected column (the shape of the Trino
`NonProjectionCompatibleRankMerger` fixture) -- happy to add one if you think
it is worth the fixture.
--
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]
