Skip to content

fix: support reading tables partitioned on a nested column #19123

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
vinishjail97 wants to merge 3 commits into
base: master
Choose a base branch
from
+150 −5
Open

fix: support reading tables partitioned on a nested column #19123

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
49ab365
fix: support reading tables partitioned on a nested column
vinishjail97 Jun 30, 2026
30b1874
Address review feedback: add tests for nested partition column read, …
vinishjail97 Jun 30, 2026
8e064fa
ci: retrigger build (flaky runner OOM/loss on scala-2.13 scala-other-…
vinishjail97 Jun 30, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 21 additions & 4 deletions ...apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala
View file
Open in desktop

@hudi-agent hudi-agent Jul 1, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🤖 Line 429: Following up on the earlier thread — I don't think this call site passes the full partitionSchema. At line 341 appendPartitionAndProject is invoked with remainingPartitionSchema as its partitionSchema arg, so here partitionSchema is the remaining schema, whereas readBaseFile (line 532) passes the full one. With a mix of one top-level mandatory field (read from file) + one nested field (appended), e.g. partition country,nested_record.level both ordering fields: remaining=[nested_record.level] (len 1), indices={1}, partitionValues.numFields=2 → line 424 is false → getFixedPartitionValues(partitionValues, remaining, {1}) does toSeq(remaining) which materializes only ordinal 0 (country's value, as StringType) then filters to {1} → empty, so the nested value is dropped. The new tests are COW-only so they hit readBaseFile (correct) and never this MOR branch. Could you confirm whether this should pass the full partitionSchema to getFixedPartitionValues here (while keeping remainingPartitionSchema for the join projection)?

⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.

Original file line number Diff line number Diff line change
Expand Up @@ -210,7 +210,12 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath: String,
}
originalVectorTypes.map {
o: Seq[String] => o.zipWithIndex.map(a => {
if (a._2 >= requiredSchema.length && mandatoryFields.contains(partitionSchema.fields(a._2 - requiredSchema.length).name)) {
val isPartitionField = a._2 >= requiredSchema.length
if (isPartitionField
&& {
val fieldName = partitionSchema.fields(a._2 - requiredSchema.length).name
mandatoryFields.contains(fieldName) && !isNestedPartitionField(fieldName)
}) {
regularVectorType
} else {
a._1
Expand Down Expand Up @@ -253,7 +258,11 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath: String,
val augmentedStorageConf = new HadoopStorageConfiguration(hadoopConf).getInline
setSchemaEvolutionConfigs(augmentedStorageConf)
augmentedStorageConf.set(ENABLE_LOGICAL_TIMESTAMP_REPAIR, hasTimestampMillisFieldInTableSchema.toString)
val (remainingPartitionSchemaArr, fixedPartitionIndexesArr) = partitionSchema.fields.toSeq.zipWithIndex.filter(p => !mandatoryFields.contains(p._1.name)).unzip
// Nested partition columns (e.g. "nested_record.level") are never read from the data file: the
// flattened dotted name is not a valid top-level field and the value is materialized from the
// partition path. Always keep them in the appended ("remaining") partition fields so they are
// not converted into a top-level Avro field below, which would fail Avro name validation.
val (remainingPartitionSchemaArr, fixedPartitionIndexesArr) = partitionSchema.fields.toSeq.zipWithIndex.filter(p => !mandatoryFields.contains(p._1.name) || isNestedPartitionField(p._1.name)).unzip

@hudi-agent hudi-agent Jun 30, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🤖 Now that a nested partition field can be appended (remaining) while a top-level partition field is still read from the file, the snapshot/MOR path hits appendPartitionAndProject, which calls getFixedPartitionValues(partitionValues, remainingPartitionSchema, fixedPartitionIndexes) (line 426). But fixedPartitionIndexes are positions in the full partitionSchema, and getFixedPartitionValues does allPartitionValues.toSeq(partitionSchema) — the base-file path correctly passes the full schema (line 529), whereas here it gets remainingPartitionSchema. Could a nested partition field mixed with a top-level field that's read from the file (e.g. timestamp keygen over multiple partition cols) drop/misread the nested value here? @nsivabalan could you verify this path?

⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.

@vinishjail97 vinishjail97 Jun 30, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Both getFixedPartitionValues call sites pass the full partitionSchema, not remainingPartitionSchema: appendPartitionAndProject (line ~426) and readBaseFile (line ~529) both call getFixedPartitionValues(<values>, partitionSchema, fixedPartitionIndexes). fixedPartitionIndexes are positions in that same full partitionSchema, so allPartitionValues.toSeq(partitionSchema) and the index set line up — the nested value is selected by its true position.

To cover exactly the mixed scenario you describe (a top-level partition column read from the file + a nested partition column appended from the path), I added testReadTablePartitionedOnNestedAndTopLevelColumns in 30b1874 (partition country,nested_record.level, both ordering fields). It hits the readBaseFile branch where remainingPartitionSchema (the nested field) differs from the full partitionSchema, and asserts country (from file) and nested_record.level (from path) are both correct. Verified on Spark 3.5: fails with Illegal character in: nested_record.level without the fix, passes with it. @nsivabalan PTAL.


// The schema of the partition cols we want to append the value instead of reading from the file
val remainingPartitionSchema = StructType(remainingPartitionSchemaArr)
Expand All @@ -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)))

@hudi-agent hudi-agent Jun 30, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🤖 A partition field lands in mandatoryFields precisely because its partition-path encoding isn't the true column value (precombine, or variable timestamp/custom keygen), so it's read from the file. By excluding nested mandatory fields here and appending them from the partition path instead, do we risk the output value being the path-encoded form (or null on a type mismatch) for those keygen cases? The merge precombine value can still come from the nested struct read via the data schema, but the appended flat partition column value would now be path-derived — is that intended?

⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.

@vinishjail97 vinishjail97 Jun 30, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good question, but there is no regression here for a nested partition field specifically. A nested partition column (nested_record.level) is never a flat physical column in the data file — the file persists the nested_record struct, not a top-level nested_record.level column — so its value can only ever be materialized from the partition path. Before this change the read did not produce a path-derived-vs-true-value discrepancy; it failed outright with Illegal character in: nested_record.level in convertStructTypeToHoodieSchema, i.e. the table was unreadable. Appending from the path is therefore the only available (and correct) source, and it is consistent with how _partitionSchemaFromProperties already treats TimestampBasedKeyGenerator partitions (StringType materialized from the path).

The precombine/merge value is unaffected: it is read from the nested_record struct via the data schema, not from the synthetic flat partition column. The fix only changes the appended output partition column (which equals the path value), not what merging sees.

Note the guard is scoped to nested (dotted) names — top-level mandatory partition columns (incl. variable timestamp/custom-keygen columns) are still read from the file exactly as before. Covered by the two regression tests added in 30b1874.

@vinishjail97 vinishjail97 Jul 1, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@yihua Have a question to clarify here - required nested partitioning fields will be read via the data file or will be parsed from the partition path? That's the bug this PR is trying to fix.

@vinishjail97 vinishjail97 Jul 1, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.:

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.

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)))

@vinishjail97 vinishjail97 Jul 1, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

val dataSchema = HoodieSchemaUtils.pruneDataSchema(schema, HoodieSchemaConversionUtils.convertStructTypeToHoodieSchema(dataStructTypeWithMandatoryPartitionFields, sanitizedTableName), exclusionFields)

spark.sessionState.conf.setConfString("spark.sql.parquet.enableVectorizedReader", supportVectorizedRead.toString)
Expand Down Expand Up @@ -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(".")

@vinishjail97 vinishjail97 Jul 1, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.


private def getFixedPartitionValues(allPartitionValues: InternalRow, partitionSchema: StructType, fixedPartitionIndexes: Set[Int]): InternalRow = {
InternalRow.fromSeq(allPartitionValues.toSeq(partitionSchema).zipWithIndex.filter(p => fixedPartitionIndexes.contains(p._2)).map(p => p._1))
}
Expand Down
130 changes: 129 additions & 1 deletion ...-spark/src/test/scala/org/apache/hudi/functional/TestFileGroupReaderPartitionColumn.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@ package org.apache.hudi.functional
import org.apache.hudi.testutils.SparkClientFunctionalTestHarness

import org.apache.spark.sql.{Row, SaveMode}
import org.apache.spark.sql.types.{DoubleType, LongType, StringType, StructField, StructType}
import org.apache.spark.sql.types.{DoubleType, IntegerType, LongType, StringType, StructField, StructType}
import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertNotNull}
import org.junit.jupiter.api.Test

Expand Down Expand Up @@ -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

@vinishjail97 vinishjail97 Jul 1, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

def testReadTablePartitionedOnNestedColumnThatIsAlsoPrecombine(): Unit = {
val nestedSchema = StructType(Array(
StructField("nested_int", IntegerType, nullable = false),
StructField("level", StringType, nullable = false)
))
val schema = StructType(Array(
StructField("id", LongType, nullable = false),
StructField("name", StringType, nullable = true),
StructField("nested_record", nestedSchema, nullable = true)
))

val opts = Map(
"hoodie.table.name" -> "test_nested_partition_precombine",
"hoodie.datasource.write.table.type" -> "COPY_ON_WRITE",
"hoodie.datasource.write.recordkey.field" -> "id",
// partition column is ALSO the precombine field -> it becomes mandatory (read from file),
// which is what drives the nested-name conversion the fix guards against.
"hoodie.datasource.write.partitionpath.field" -> "nested_record.level",
"hoodie.datasource.write.precombine.field" -> "nested_record.level",
"hoodie.insert.shuffle.parallelism" -> "2",
"hoodie.upsert.shuffle.parallelism" -> "2"
)

val rows = Seq(
Row(1L, "a", Row(10, "INFO")),
Row(2L, "b", Row(20, "ERROR")),
Row(3L, "c", Row(30, "INFO")),
Row(4L, "d", Row(40, "DEBUG"))
)
spark.createDataFrame(spark.sparkContext.parallelize(rows, 1), schema)
.write.format("hudi")
.options(opts)
.mode(SaveMode.Overwrite)
.save(basePath)

// Full read: the nested partition column must materialize (from the partition path) for every
// row rather than crashing the scan.
val all = spark.read.format("hudi").load(basePath)
.select("id", "nested_record")
.collect()
.map(r => r.getLong(0) -> r.getStruct(1).getString(1))
.toMap
assertEquals(4, all.size, "all rows must read back")
assertEquals("INFO", all(1L))
assertEquals("ERROR", all(2L))
assertEquals("INFO", all(3L))
assertEquals("DEBUG", all(4L))

// Filtered read on the nested partition column (partition pruning path).
val infoIds = spark.read.format("hudi").load(basePath)
.filter("nested_record.level = 'INFO'")
.select("id")
.collect()
.map(_.getLong(0))
.toSet
assertEquals(Set(1L, 3L), infoIds, "filter on nested partition column must return INFO rows")
}

/**
* Mixed case: one top-level partition column read from the file and one nested partition column
* appended from the path, both mandatory (here via multiple ordering fields). Exercises the
* `readBaseFile` branch where `remainingPartitionSchema` (the appended nested field) differs from
* the full `partitionSchema`, confirming the nested value is materialized from the path while the
* top-level value is still read from the file.
*/
@Test
def testReadTablePartitionedOnNestedAndTopLevelColumns(): Unit = {
val nestedSchema = StructType(Array(
StructField("nested_int", IntegerType, nullable = false),
StructField("level", StringType, nullable = false)
))
val schema = StructType(Array(
StructField("id", LongType, nullable = false),
StructField("country", StringType, nullable = false),
StructField("nested_record", nestedSchema, nullable = true)
))

val opts = Map(
"hoodie.table.name" -> "test_nested_and_toplevel_partition",
"hoodie.datasource.write.table.type" -> "COPY_ON_WRITE",
"hoodie.datasource.write.recordkey.field" -> "id",
"hoodie.datasource.write.partitionpath.field" -> "country,nested_record.level",
"hoodie.datasource.write.keygenerator.class" -> "org.apache.hudi.keygen.ComplexKeyGenerator",
// both partition columns are ordering fields -> both mandatory; country is read from the file,
// the nested column is appended from the path.
"hoodie.table.ordering.fields" -> "country,nested_record.level",
"hoodie.datasource.write.hive_style_partitioning" -> "true",
"hoodie.insert.shuffle.parallelism" -> "2",
"hoodie.upsert.shuffle.parallelism" -> "2"
)

val rows = Seq(
Row(1L, "US", Row(10, "INFO")),
Row(2L, "IN", Row(20, "ERROR")),
Row(3L, "US", Row(30, "DEBUG"))
)
spark.createDataFrame(spark.sparkContext.parallelize(rows, 1), schema)
.write.format("hudi")
.options(opts)
.mode(SaveMode.Overwrite)
.save(basePath)

val all = spark.read.format("hudi").load(basePath)
.select("id", "country", "nested_record")
.collect()
.map(r => r.getLong(0) -> (r.getString(1), r.getStruct(2).getString(1)))
.toMap
assertEquals(3, all.size, "all rows must read back")
assertEquals(("US", "INFO"), all(1L))
assertEquals(("IN", "ERROR"), all(2L))
assertEquals(("US", "DEBUG"), all(3L))
}
}
Loading