deepakpanda93 commented on code in PR #19471:
URL: https://github.com/apache/hudi/pull/19471#discussion_r3712056138
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestPartitionBucketIndexSupport.scala:
##########
@@ -173,32 +172,31 @@ class TestPartitionBucketIndexSupport extends
TestBucketIndexSupport {
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 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
Review Comment:
Not ignoring it — the drift argument is the convincing bit. `76bb580ae149`
updating two of three copies and guarding the third is precisely how this suite
ended up skipped on Spark 4, so leaving a third copy in place would be
repeating the setup for the same failure.
Taken your snippet nearly verbatim into `TestBucketIndexSupport`:
```scala
protected def optimizeResolvedExpr(exprRaw: String): Expression = {
val resolveExpr = HoodieCatalystExpressionUtils.resolveExpr(spark,
exprRaw, structSchema)
val holder = HoodieDummyExpressionHolder(Seq(resolveExpr),
resolveExpr.references.toSeq)
spark.sessionState.optimizer.execute(holder).asInstanceOf[HoodieDummyExpressionHolder].exprs.head
}
```
All three call sites now collapse to `val optimizerExpr =
optimizeResolvedExpr(exprRaw)`, and
`HoodieDummyExpressionHolder(Seq(resolveExpr), ...)` appears once in the
codebase. Added a scaladoc noting the SPARK-44219 reason, so the next Spark
bump has the context at the one edit site.
This does mean the diff now touches the parent class as well as the
subclass. Given the extraction has to live there to collapse the copies, and
the duplicated assertion you flagged is at
`TestBucketIndexSupport.scala:203-205`, that seemed in scope — say the word if
you would rather keep this PR to the single file and split the parent changes
out.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestPartitionBucketIndexSupport.scala:
##########
@@ -171,34 +169,84 @@ 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 = {
Review Comment:
Half right, and the comment was imprecise — reworded, though not quite to
the suggested text.
`Set[String]` iteration order is hash-dependent, but it is not
non-deterministic: `String.hashCode` is stable across JVM runs, so a set of
fixed file names would iterate the same way every time and the split would be
reproducible. What actually varies run to run is the input, because
`BucketIdentifier.newBucketFileIdPrefix` calls `FSUtils.createNewFileIdPfx()` =
`UUID.randomUUID()`, so the names differ on every run and their hashes with
them.
Hash-ordered iteration is the mechanism; the uuid is the cause. The comment
now names both:
```scala
// 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.
```
--
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]
