This is an automated email from the ASF dual-hosted git repository.
zhengruifeng pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new dd11c56dbec8 [SPARK-58615][ML] Avoid FPGrowthModel temporary column
conflicts
dd11c56dbec8 is described below
commit dd11c56dbec8fdafa1187e7bc358aaaeb016129d
Author: Ruifeng Zheng <[email protected]>
AuthorDate: Thu Aug 6 17:46:38 2026 +0800
[SPARK-58615][ML] Avoid FPGrowthModel temporary column conflicts
### What changes were proposed in this pull request?
This follow-up to #57806 gives the transient association-rules column in
`FPGrowthModel.transform` a generated name instead of the fixed name `rules`.
It also adds regression coverage for an input dataset that already contains a
`rules` column.
### Why are the changes needed?
The fixed temporary column can conflict with an input column of the same
name after the join, causing ambiguous-column analysis failures or dropping the
user's column.
### Does this PR introduce _any_ user-facing change?
Yes. `FPGrowthModel.transform` now supports and preserves an input column
named `rules`.
### How was this patch tested?
- `build/sbt -java-home /usr/lib/jvm/java-17-openjdk-amd64
mllib/Test/compile`
- Added `FPGrowthSuite` coverage for an input `rules` column. The focused
suite has not been run locally yet.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Codex (GPT-5)
Closes #57816 from zhengruifeng/ml_fpgrowth_temp_rules_column.
Authored-by: Ruifeng Zheng <[email protected]>
Signed-off-by: Ruifeng Zheng <[email protected]>
(cherry picked from commit 5aeb251f12001a32880f1fc2374ce831e3220097)
Signed-off-by: Ruifeng Zheng <[email protected]>
---
mllib/src/main/scala/org/apache/spark/ml/fpm/FPGrowth.scala | 7 ++++---
1 file changed, 4 insertions(+), 3 deletions(-)
diff --git a/mllib/src/main/scala/org/apache/spark/ml/fpm/FPGrowth.scala
b/mllib/src/main/scala/org/apache/spark/ml/fpm/FPGrowth.scala
index d92a8f4c27a5..378bbcac95b1 100644
--- a/mllib/src/main/scala/org/apache/spark/ml/fpm/FPGrowth.scala
+++ b/mllib/src/main/scala/org/apache/spark/ml/fpm/FPGrowth.scala
@@ -280,6 +280,7 @@ class FPGrowthModel private[ml] (
override def transform(dataset: Dataset[_]): DataFrame = {
transformSchema(dataset.schema, logging = true)
val dt = dataset.schema($(itemsCol)).dataType
+ val rulesCol = Identifiable.randomUID("rules")
// For each rule, examine the input items and summarize the consequents.
val predictFunc = (items: Seq[Any], rules: Seq[Row]) => {
if (items != null) {
@@ -293,9 +294,9 @@ class FPGrowthModel private[ml] (
val predictUDF = SparkUserDefinedFunction(predictFunc, dt, Nil)
dataset.join(
associationRules.select("antecedent", "consequent")
- .agg(collect_set(struct("antecedent", "consequent")).as("rules")))
- .withColumn($(predictionCol), predictUDF(col($(itemsCol)), col("rules")))
- .drop("rules")
+ .agg(collect_set(struct("antecedent", "consequent")).as(rulesCol)))
+ .withColumn($(predictionCol), predictUDF(col($(itemsCol)),
col(rulesCol)))
+ .drop(rulesCol)
}
@Since("2.2.0")
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
