vinishjail97 commented on code in PR #19123:
URL: https://github.com/apache/hudi/pull/19123#discussion_r3502481199
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala:
##########
@@ -265,9 +274,9 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
val exclusionFields = new java.util.HashSet[String]()
exclusionFields.add("op")
partitionSchema.fields.foreach(f => exclusionFields.add(f.name))
- val requestedStructType = StructType(requiredSchema.fields ++
partitionSchema.fields.filter(f => mandatoryFields.contains(f.name)))
+ val requestedStructType = StructType(requiredSchema.fields ++
partitionSchema.fields.filter(f => mandatoryFields.contains(f.name) &&
!isNestedPartitionField(f.name)))
val requestedSchema = HoodieSchemaUtils.pruneDataSchema(schema,
HoodieSchemaConversionUtils.convertStructTypeToHoodieSchema(requestedStructType,
sanitizedTableName), exclusionFields)
- val dataStructTypeWithMandatoryPartitionFields =
StructType(dataStructType.fields ++ partitionSchema.fields.filter(f =>
mandatoryFields.contains(f.name)))
+ val dataStructTypeWithMandatoryPartitionFields =
StructType(dataStructType.fields ++ partitionSchema.fields.filter(f =>
mandatoryFields.contains(f.name) && !isNestedPartitionField(f.name)))
Review Comment:
When the nested partition column is also the precombine/ordering field, it's
now excluded from the `dataSchema`/`requestedSchema` passed to
`HoodieFileGroupReader`. For MOR merges the ordering field is what resolves
which record wins — is it safe for the reader to never receive it in this case?
Since a partition column is constant across all records in a file group the
tiebreak is effectively a no-op, so I suspect this is fine, but it'd be good to
confirm and cover it with the MOR test suggested above.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestFileGroupReaderPartitionColumn.scala:
##########
@@ -155,4 +155,132 @@ class TestFileGroupReaderPartitionColumn extends
SparkClientFunctionalTestHarnes
assertEquals("IN", rows(12L), "id=12 partition column must be IN")
assertEquals("IN", rows(14L), "id=14 partition column must be IN")
}
+
+ /**
+ * Regression test for reading a table partitioned on a nested column when
that nested column is
+ * also a mandatory field (here, the precombine/ordering field). Being
mandatory, the file group
+ * reader requests the partition column as a top-level field; for a nested
path
+ * ("nested_record.level") this previously failed in
`buildReaderWithPartitionValues` with
+ * `HoodieSchemaException: Illegal character in: nested_record.level` when
converting the
+ * StructType into an Avro-backed HoodieSchema.
+ *
+ * The nested partition value is never a flat top-level column in the data
file — it must be
+ * materialized from the partition path — so the fix keeps it out of the
file-read schema and
+ * appends it from the path. The existing
`TestCOWDataSource#testNestedFieldPartition` covers the
+ * common (non-mandatory) path where the field is already appended from the
path and does not hit
+ * this case.
+ */
+ @Test
Review Comment:
Both new tests are COPY_ON_WRITE and only exercise the `readBaseFile` code
path. The snapshot-with-log-files branch (`appendPartitionAndProject`, ~L341 in
the file format) also consumes
`remainingPartitionSchema`/`fixedPartitionIndexes` for a nested mandatory field
and isn't covered here. Could you add a MERGE_ON_READ variant that writes an
initial batch then an update batch (so log files exist) on a table partitioned
on a nested mandatory column, and read it back? That would lock down the
log-merge path this fix also touches. Non-blocking but valuable.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala:
##########
@@ -265,9 +274,9 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
val exclusionFields = new java.util.HashSet[String]()
exclusionFields.add("op")
partitionSchema.fields.foreach(f => exclusionFields.add(f.name))
- val requestedStructType = StructType(requiredSchema.fields ++
partitionSchema.fields.filter(f => mandatoryFields.contains(f.name)))
+ val requestedStructType = StructType(requiredSchema.fields ++
partitionSchema.fields.filter(f => mandatoryFields.contains(f.name) &&
!isNestedPartitionField(f.name)))
Review Comment:
The "mandatory AND not-nested" (i.e. read-from-file) predicate is now
duplicated here, at L279, and in `vectorTypes` (L217), and appears De
Morgan-negated in the `remainingPartitionSchema` filter (L265). Consider
extracting a single helper, e.g.:
```scala
private def partitionFieldReadFromFile(field: StructField): Boolean =
mandatoryFields.contains(field.name) && !isNestedPartitionField(field.name)
```
and reusing it in all four spots. That keeps the negated form at L265 from
silently drifting out of sync with the positive forms if the condition ever
changes. Non-blocking.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala:
##########
@@ -548,6 +557,14 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
}.asInstanceOf[Iterator[InternalRow]]
}
+ /**
+ * A partition column whose name is a nested field path (e.g.
"nested_record.level") cannot be
+ * read from the data file as a flat top-level column, nor converted into a
top-level Avro field
+ * (Avro rejects '.' in names). Its value is always materialized from the
partition path, so such
+ * fields are treated as appended partition fields rather than read from the
file.
+ */
+ private def isNestedPartitionField(name: String): Boolean =
name.contains(".")
Review Comment:
nit: `name.contains(".")` is a sound heuristic here — top-level Avro/Spark
field names can't contain `.`, so there are no false positives, and the javadoc
explains it well. Minor: the key generators already split `partitionpath.field`
on `.` to express nesting; if there's a shared constant/util for that
separator, reusing it would be preferable to a bare literal.
--
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]
