Re: [PR] fix: support reading tables partitioned on a nested column [hudi]

["vinishjail97 (via GitHub)" <gi...@apache.org>]

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: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org