This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new c6662e742d21 test(spark): stop skipping
TestPartitionBucketIndexSupport on Spark 4 (#19471)
c6662e742d21 is described below
commit c6662e742d21358633adb66c7be710b772307479
Author: deepakpanda93 <[email protected]>
AuthorDate: Fri Aug 14 14:16:04 2026 +0530
test(spark): stop skipping TestPartitionBucketIndexSupport on Spark 4
(#19471)
* test(spark): stop skipping TestPartitionBucketIndexSupport on Spark 4
exprFilePathAnswerCheck built its plan with Spark's DummyExpressionHolder,
which
hardcodes output to Nil. The attributes referenced by the expression are
then
dangling, and the validation rule added by SPARK-44219 rejects the plan:
[PLAN_VALIDATION_FAILED_RULE_EXECUTOR] The input plan of
BaseSessionStateBuilder$$anon$2 is invalid: Aliases A#0L are dangling in
the
references for plan: DummyExpressionHolder [(A#0L = cast(3 as bigint))]
The method was guarded with !gteqSpark4_0, so on Spark 4 it returned before
asserting anything and the eleven checks across
testSingleHashFieldsExpression
and testMultipleHashFieldsExpress passed vacuously.
Reuse HoodieDummyExpressionHolder, which TestBucketIndexSupport already
forked
for this exact reason and which takes the output explicitly, and drop the
guard
so the assertions run on every Spark version.
Reproducing the failure needs spark.testing to be set, since Spark gates
plan
change validation behind Utils.isTesting. A full module run sets it via
HoodieSparkSqlTestBase's static initializer, which is why it surfaces in CI
but
not when the test is run on its own.
* test(spark): address review on the bucket index support tests
- Build the second partition's file slices with partitionPath2 rather than
partitionPath1. Inert today, since the candidate computation takes the
partition from the tuple, but a trap for the next change.
- Order the file names before splitting them across partitions. The file id
prefix carries a uuid, so which bucket landed in which partition varied
from run to run.
- Cover per partition bucket counts. Every partition the existing assertions
use matches the expression and so carries the same bucket count, which
leaves the mapping itself untested; the new case pairs a matching
partition
with one that falls back to the table default and includes a file that
would only be a candidate under the wrong count.
- Lift the expression holder setup into the parent as optimizeResolvedExpr.
It was copied in three places, and one copy going stale is what left this
suite skipped on Spark 4.
- Drop the assertion that repeated "A = 5 And (A = 2 Or B = 'abc')" verbatim
in both classes.
---
.../hudi/functional/TestBucketIndexSupport.scala | 25 +++--
.../TestPartitionBucketIndexSupport.scala | 113 +++++++++++++++------
2 files changed, 95 insertions(+), 43 deletions(-)
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestBucketIndexSupport.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestBucketIndexSupport.scala
index 7a4f575d0b27..cc2a040f8612 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestBucketIndexSupport.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestBucketIndexSupport.scala
@@ -200,9 +200,6 @@ class TestBucketIndexSupport extends
HoodieSparkClientTestBase with PredicateHel
equalTo = "A = 5 And (A = 2 Or B = 'abc')"
exprBucketAnswerCheck(bucketIndexSupport, equalTo, List.apply(bucket5Id8),
fallback = false)
exprFilePathAnswerCheck(bucketIndexSupport, equalTo,
Set.apply(bucket5Id8FileName), allFileNames, fallback = false)
- equalTo = "A = 5 And (A = 2 Or B = 'abc')"
- exprBucketAnswerCheck(bucketIndexSupport, equalTo, List.apply(bucket5Id8),
fallback = false)
- exprFilePathAnswerCheck(bucketIndexSupport, equalTo,
Set.apply(bucket5Id8FileName), allFileNames, fallback = false)
var inExpr = "A in (3)"
exprBucketAnswerCheck(bucketIndexSupport, inExpr, List.apply(bucket3Id6),
fallback = false)
@@ -379,11 +376,20 @@ class TestBucketIndexSupport extends
HoodieSparkClientTestBase with PredicateHel
exprBucketAnswerCheck(bucketIndexSupport, fallBack, List.empty, fallback =
true)
}
- def exprBucketAnswerCheck(bucketIndexSupport: BucketIndexSupport, exprRaw:
String, expectResult: List[Int], fallback: Boolean): Unit = {
+ /**
+ * Resolves an expression and runs it through the optimizer, wrapped in
+ * [[HoodieDummyExpressionHolder]] so that the expression's references are
exposed as plan output.
+ * Spark 4 validates plans after every optimizer rule (SPARK-44219) and
rejects a plan whose
+ * aliases are not reachable from its output, which the holder Spark ships
does not satisfy.
+ */
+ protected def optimizeResolvedExpr(exprRaw: String): Expression = {
val resolveExpr = HoodieCatalystExpressionUtils.resolveExpr(spark,
exprRaw, structSchema)
- val dummyExpressionHolder = HoodieDummyExpressionHolder(Seq(resolveExpr),
resolveExpr.references.toSeq)
- val optimizerPlan =
spark.sessionState.optimizer.execute(dummyExpressionHolder)
- val optimizerExpr =
optimizerPlan.asInstanceOf[HoodieDummyExpressionHolder].exprs.head
+ val holder = HoodieDummyExpressionHolder(Seq(resolveExpr),
resolveExpr.references.toSeq)
+
spark.sessionState.optimizer.execute(holder).asInstanceOf[HoodieDummyExpressionHolder].exprs.head
+ }
+
+ def exprBucketAnswerCheck(bucketIndexSupport: BucketIndexSupport, exprRaw:
String, expectResult: List[Int], fallback: Boolean): Unit = {
+ val optimizerExpr = optimizeResolvedExpr(exprRaw)
val bucketSet =
bucketIndexSupport.filterQueriesWithBucketHashField(splitConjunctivePredicates(optimizerExpr))
if (fallback) {
@@ -402,10 +408,7 @@ class TestBucketIndexSupport extends
HoodieSparkClientTestBase with PredicateHel
def exprFilePathAnswerCheck(bucketIndexSupport: BucketIndexSupport, exprRaw:
String, expectResult: Set[String],
allFileStatus: Set[String], fallback: Boolean):
Unit = {
- val resolveExpr = HoodieCatalystExpressionUtils.resolveExpr(spark,
exprRaw, structSchema)
- val dummyExpressionHolder = HoodieDummyExpressionHolder(Seq(resolveExpr),
resolveExpr.references.toSeq)
- val optimizerPlan =
spark.sessionState.optimizer.execute(dummyExpressionHolder)
- val optimizerExpr =
optimizerPlan.asInstanceOf[HoodieDummyExpressionHolder].exprs.head
+ val optimizerExpr = optimizeResolvedExpr(exprRaw)
val bucketSet =
bucketIndexSupport.filterQueriesWithBucketHashField(splitConjunctivePredicates(optimizerExpr))
if (fallback) {
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestPartitionBucketIndexSupport.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestPartitionBucketIndexSupport.scala
index cd3984d074fd..944f4c34e952 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestPartitionBucketIndexSupport.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestPartitionBucketIndexSupport.scala
@@ -17,7 +17,7 @@
package org.apache.hudi.functional
-import org.apache.hudi.{HoodieFileIndex, HoodieSparkUtils,
PartitionBucketIndexSupport}
+import org.apache.hudi.{HoodieFileIndex, PartitionBucketIndexSupport}
import org.apache.hudi.common.config.{HoodieMetadataConfig, TypedProperties}
import org.apache.hudi.common.fs.FSUtils
import org.apache.hudi.common.model.{FileSlice, HoodieBaseFile,
PartitionBucketIndexHashingConfig}
@@ -30,8 +30,6 @@ import org.apache.hudi.keygen.constant.KeyGeneratorOptions
import org.apache.hudi.storage.{StoragePath, StoragePathInfo}
import org.apache.avro.generic.GenericData
-import org.apache.spark.sql.HoodieCatalystExpressionUtils
-import org.apache.spark.sql.catalyst.encoders.DummyExpressionHolder
import org.junit.jupiter.api.{BeforeEach, Tag, Test}
import org.mockito.Mockito
@@ -43,6 +41,8 @@ class TestPartitionBucketIndexSupport extends
TestBucketIndexSupport {
private val DEFAULT_EXPRESSIONS =
"\\d{4}\\-(06\\-(01|17|18)|11\\-(01|10|11))," + EXPRESSION_BUCKET_NUMBER
private val DEFAULT_BUCKET_NUMBER = 10
private val DEFAULT_PARTITION_PATH = Array("2025-06-17", "2025-06-18")
+ // deliberately outside DEFAULT_EXPRESSIONS, so it hashes into
DEFAULT_BUCKET_NUMBER buckets
+ private val NON_MATCHING_PARTITION_PATH = "2025-07-01"
private var fileIndex: HoodieFileIndex = null
@BeforeEach
override def setUp(): Unit = {
@@ -111,8 +111,6 @@ class TestPartitionBucketIndexSupport extends
TestBucketIndexSupport {
exprFilePathAnswerCheck(bucketIndexSupport, equalTo,
Set.apply(bucket5Id8FileName, bucket2Id5FileName), allFileNames)
equalTo = "A = 5 And (A = 2 Or B = 'abc')"
exprFilePathAnswerCheck(bucketIndexSupport, equalTo,
Set.apply(bucket5Id8FileName), allFileNames)
- equalTo = "A = 5 And (A = 2 Or B = 'abc')"
- exprFilePathAnswerCheck(bucketIndexSupport, equalTo,
Set.apply(bucket5Id8FileName), allFileNames)
}
@Test
@@ -171,34 +169,85 @@ class TestPartitionBucketIndexSupport extends
TestBucketIndexSupport {
exprFilePathAnswerCheck(bucketIndexSupport, equalTo,
Set.apply(bucket4Id7FileName), allFileNames)
}
+ /**
+ * Every partition used by the other tests matches the expression, so all of
them carry the same
+ * bucket count and nothing there notices if a partition is hashed with the
wrong one. Pair a
+ * partition the expression matches, which gets [[EXPRESSION_BUCKET_NUMBER]]
buckets, with one it
+ * does not, which falls back to the table default, and check that a file is
only a candidate
+ * under the bucket count belonging to its own partition.
+ */
+ @Test
+ def testCandidateFilesUsePerPartitionBucketCount(): Unit = {
+ val configProperties = new TypedProperties()
+
configProperties.setProperty(HoodieIndexConfig.BUCKET_INDEX_HASH_FIELD.key, "A")
+ configProperties.setProperty(HoodieTableConfig.RECORDKEY_FIELDS.key, "A")
+ configProperties.setProperty(KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key,
"A")
+
configProperties.setProperty(HoodieIndexConfig.BUCKET_INDEX_NUM_BUCKETS.key,
String.valueOf(DEFAULT_BUCKET_NUMBER))
+ metaClient.getTableConfig.setValue(HoodieTableConfig.CREATE_SCHEMA.key(),
avroSchemaStr)
+ val metadataConfig = HoodieMetadataConfig.newBuilder
+ .fromProperties(configProperties)
+ .enable(configProperties.getBoolean(HoodieMetadataConfig.ENABLE.key,
true)).build()
+ val bucketIndexSupport = new PartitionBucketIndexSupport(spark,
metadataConfig, metaClient)
+
+ val record = new GenericData.Record(schema.toAvroSchema)
+ record.put("A", "3")
+ val recordKey = new
NonpartitionedKeyGenerator(configProperties).getKey(record).getRecordKey
+ val bucketIdInExpressionPartition =
BucketIdentifier.getBucketId(recordKey, "A", EXPRESSION_BUCKET_NUMBER)
+ val bucketIdInDefaultPartition = BucketIdentifier.getBucketId(recordKey,
"A", DEFAULT_BUCKET_NUMBER)
+ // the two bucket counts have to disagree for this record, otherwise the
check below proves nothing
+ assert(bucketIdInExpressionPartition != bucketIdInDefaultPartition)
+
+ // the file id prefix carries a fresh uuid every call, so the same bucket
can be named twice
+ def fileNameForBucket(bucketId: Int): String =
FSUtils.makeBaseFileName("00000000000000000",
+ FSUtils.makeWriteToken(1, 0, 1),
BucketIdentifier.newBucketFileIdPrefix(bucketId) + "-0",
+ HoodieTableConfig.BASE_FILE_FORMAT.defaultValue.getFileExtension)
+
+ val expressionPartition = DEFAULT_PARTITION_PATH(0)
+ val defaultPartition = NON_MATCHING_PARTITION_PATH
+ val expectedFromExpressionPartition =
fileNameForBucket(bucketIdInExpressionPartition)
+ val expectedFromDefaultPartition =
fileNameForBucket(bucketIdInDefaultPartition)
+ // same bucket as the matching partition's candidate, but sitting in the
partition that hashes
+ // into the table default, so it must not be picked up
+ val decoyInDefaultPartition =
fileNameForBucket(bucketIdInExpressionPartition)
+
+ val input = Seq(
+ (Option.apply(new
BaseHoodieTableFileIndex.PartitionPath(expressionPartition, Array())),
+ fileSlicesOf(expressionPartition,
Seq(expectedFromExpressionPartition))),
+ (Option.apply(new
BaseHoodieTableFileIndex.PartitionPath(defaultPartition, Array())),
+ fileSlicesOf(defaultPartition, Seq(expectedFromDefaultPartition,
decoyInDefaultPartition))))
+
+ val candidate = bucketIndexSupport.computeCandidateFileNames(fileIndex,
+ splitConjunctivePredicates(optimizeResolvedExpr("A = 3")), Seq(), input,
false)
+
+ assert(candidate.get.equals(Set.apply(expectedFromExpressionPartition,
expectedFromDefaultPartition)))
+ }
+
def exprFilePathAnswerCheck(bucketIndexSupport: PartitionBucketIndexSupport,
exprRaw: String, expectResult: Set[String],
allFileStatus: Set[String]): Unit = {
- if (!HoodieSparkUtils.gteqSpark4_0) { // TODO (HUDI-9403)
- val resolveExpr = HoodieCatalystExpressionUtils.resolveExpr(spark,
exprRaw, structSchema)
- val optimizerPlan =
spark.sessionState.optimizer.execute(DummyExpressionHolder(Seq(resolveExpr)))
- val optimizerExpr =
optimizerPlan.asInstanceOf[DummyExpressionHolder].exprs.head
-
- // split input files into different partitions
- val partitionPath1 = DEFAULT_PARTITION_PATH(0)
- val allFileSlices1: Seq[FileSlice] = allFileStatus.slice(0,
3).map(fileName => {
- val slice = new FileSlice(partitionPath1, "00000000000000000",
FSUtils.getFileId(fileName))
- slice.setBaseFile(new HoodieBaseFile(new StoragePathInfo(new
StoragePath(fileName), 0L, false, 0, 0, 0)))
- slice
- }).toSeq
-
- val partitionPath2 = DEFAULT_PARTITION_PATH(1)
- val allFileSlices2: Seq[FileSlice] = allFileStatus.slice(3,
5).map(fileName => {
- val slice = new FileSlice(partitionPath1, "00000000000000000",
FSUtils.getFileId(fileName))
- slice.setBaseFile(new HoodieBaseFile(new StoragePathInfo(new
StoragePath(fileName), 0L, false, 0, 0, 0)))
- slice
- }).toSeq
-
- val input = Seq((Option.apply(new
BaseHoodieTableFileIndex.PartitionPath(partitionPath1, Array())),
allFileSlices1),
- (Option.apply(new
BaseHoodieTableFileIndex.PartitionPath(partitionPath2, Array())),
allFileSlices2))
- val candidate = bucketIndexSupport.computeCandidateFileNames(fileIndex,
splitConjunctivePredicates(optimizerExpr),
- Seq(), input, false)
-
- assert(candidate.get.equals(expectResult))
- }
+ val optimizerExpr = optimizeResolvedExpr(exprRaw)
+
+ // Split the input files across two partitions. A Set iterates in element
hash order, and these
+ // file names embed a fresh uuid on every run, so the unsorted split put
different buckets in
+ // each partition from one run to the next. Sort first to pin the file to
partition assignment.
+ val orderedFileNames = allFileStatus.toSeq.sorted
+ val partitionPath1 = DEFAULT_PARTITION_PATH(0)
+ val partitionPath2 = DEFAULT_PARTITION_PATH(1)
+ val input = Seq(
+ (Option.apply(new BaseHoodieTableFileIndex.PartitionPath(partitionPath1,
Array())),
+ fileSlicesOf(partitionPath1, orderedFileNames.slice(0, 3))),
+ (Option.apply(new BaseHoodieTableFileIndex.PartitionPath(partitionPath2,
Array())),
+ fileSlicesOf(partitionPath2, orderedFileNames.slice(3, 5))))
+ val candidate = bucketIndexSupport.computeCandidateFileNames(fileIndex,
splitConjunctivePredicates(optimizerExpr),
+ Seq(), input, false)
+
+ assert(candidate.get.equals(expectResult))
+ }
+
+ private def fileSlicesOf(partitionPath: String, fileNames: Seq[String]):
Seq[FileSlice] = {
+ fileNames.map(fileName => {
+ val slice = new FileSlice(partitionPath, "00000000000000000",
FSUtils.getFileId(fileName))
+ slice.setBaseFile(new HoodieBaseFile(new StoragePathInfo(new
StoragePath(fileName), 0L, false, 0, 0, 0)))
+ slice
+ })
}
}
