wombatu-kun commented on code in PR #19458:
URL: https://github.com/apache/hudi/pull/19458#discussion_r3701088311
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala:
##########
@@ -81,7 +80,6 @@ case class HoodieTableState(tablePath: String,
recordKeyField: String,
orderingFields: List[String],
usesVirtualKeys: Boolean,
Review Comment:
With recordPayloadClassName gone, latestCommitTimestamp is the only field of
HoodieTableState that anything still reads. Worth dropping the other seven by
the same criterion, or noting that they are deliberately kept - follow-up, not
a blocker.
##########
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:
Both assertions here read FileSourceScanExec.requiredSchema and
post-projection row values, which is what the DefaultHoodieRecordPayload
version asserted too. Does this test fail if the CUSTOM projection-incompatible
branch in FileGroupReaderSchemaHandler is removed, or is SchemaHandlerTestBase
the only place pinning it?
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieHadoopFsRelationFactory.scala:
##########
@@ -335,7 +335,7 @@ class
HoodieMergeOnReadIncrementalHadoopFsRelationFactoryV2(override val sqlCont
isBootstrap:
Boolean,
rangeType:
RangeType = RangeType.OPEN_CLOSED)
extends HoodieMergeOnReadIncrementalHadoopFsRelationFactory(sqlContext,
metaClient, options, schemaSpec, isBootstrap,
- MergeOnReadIncrementalRelationV2(sqlContext, options, metaClient,
schemaSpec, None, rangeType))
+ MergeOnReadIncrementalRelationV2(sqlContext, options, metaClient,
schemaSpec, rangeType))
Review Comment:
The description says only metadata-table reads and the schema-on-read branch
still produce a HoodieBaseRelation, but the incremental factories here build
MergeOnReadIncrementalRelationV1/V2 on the non-metadata path for both MOR and
COW. They never reach a LogicalRelation because HoodieIncrementalFileIndex only
calls listFileSplits and getRequiredFilters on them, so the conclusion stands -
could you correct the enumeration in the description?
##########
hudi-common/src/test/java/org/apache/hudi/common/table/read/SchemaHandlerTestBase.java:
##########
@@ -130,6 +130,27 @@ public void testMor(RecordMergeMode mergeMode,
}
assertEquals(expectedRequiredSchema, schemaHandler.getRequiredSchema());
assertFalse(readerContext.getNeedsBootstrapMerge());
+
+ //read subset of columns with a nested-narrowed field, the shape Spark's
nested schema pruning
+ //requests: "fare" keeps only its "amount" leaf
+ requestedSchema =
narrowFareToAmountOnly(generateProjectionSchema("begin_lat", "fare", "rider"));
+ schemaHandler = createSchemaHandler(readerContext, dataSchema,
requestedSchema, supportsParquetRowIndex);
+ if (mergeMode == EVENT_TIME_ORDERING && hasPrecombine) {
+ expectedRequiredSchema =
narrowFareToAmountOnly(generateProjectionSchema(hasBuiltInDelete, "begin_lat",
"fare", "rider", "_hoodie_record_key", "timestamp"));
+ } else if (mergeMode == EVENT_TIME_ORDERING || mergeMode ==
COMMIT_TIME_ORDERING) {
+ expectedRequiredSchema =
narrowFareToAmountOnly(generateProjectionSchema(hasBuiltInDelete, "begin_lat",
"fare", "rider", "_hoodie_record_key"));
+ } else if (mergeMode == CUSTOM && isProjectionCompatible) {
+ expectedRequiredSchema =
narrowFareToAmountOnly(generateProjectionSchema("begin_lat", "fare", "rider",
"begin_lon", "_hoodie_record_key", "timestamp"));
+ } else {
+ //a projection-incompatible custom merger may need any column to merge,
so the handler must
+ //re-expand the nested-narrowed request back to the full data schema
(see HUDI-5443)
+ expectedRequiredSchema = dataSchema;
Review Comment:
The CUSTOM projection-incompatible branch returns the table schema before it
looks at the requested schema, so narrowing fare changes nothing here and this
repeats the assertion already made for the flat projection above. Consider
dropping this branch and keeping the block for the three cases where the nested
narrowing actually has to survive.
##########
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") {
Review Comment:
The PR description says this suite is kept unchanged, but this commit
renames the test, swaps in a custom payload, enables meta fields and adds an
UPDATE plus a new answer assertion. Could you refresh the description so
Summary and Risk Level also cover the test changes and the
HoodieTableState.recordPayloadClassName removal?
--
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]
