You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2025/07/01 15:12:43 UTC
(spark) branch branch-3.5 updated: Revert "[SPARK-52023][SQL] Fix data corruption/segfault returning Option[Product] from udaf"
This is an automated email from the ASF dual-hosted git repository.
dongjoon pushed a commit to branch branch-3.5
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.5 by this push:
new a56879e67a11 Revert "[SPARK-52023][SQL] Fix data corruption/segfault returning Option[Product] from udaf"
a56879e67a11 is described below
commit a56879e67a11967b54812a8365d5bedcfc3d1861
Author: Dongjoon Hyun <do...@apache.org>
AuthorDate: Tue Jul 1 08:12:19 2025 -0700
Revert "[SPARK-52023][SQL] Fix data corruption/segfault returning Option[Product] from udaf"
This reverts commit 3f2a3ba440d9a5e6d719f739f7e026f57a8e1c32.
---
.../spark/sql/execution/aggregate/udaf.scala | 2 +-
.../spark/sql/hive/execution/UDAQuerySuite.scala | 27 ----------------------
2 files changed, 1 insertion(+), 28 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/udaf.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/udaf.scala
index fe6307b5bbe8..e517376bc5fc 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/udaf.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/udaf.scala
@@ -530,7 +530,7 @@ case class ScalaAggregator[IN, BUF, OUT](
def eval(buffer: BUF): Any = {
val row = outputSerializer(agg.finish(buffer))
- if (outputEncoder.isSerializedAsStructForTopLevel) row else row.get(0, dataType)
+ if (outputEncoder.isSerializedAsStruct) row else row.get(0, dataType)
}
private[this] lazy val bufferRow = new UnsafeRow(bufferEncoder.namedExpressions.length)
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/UDAQuerySuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/UDAQuerySuite.scala
index daa08666ea0e..0bd6b1403d39 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/UDAQuerySuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/UDAQuerySuite.scala
@@ -60,22 +60,6 @@ object LongProductSumAgg extends Aggregator[(jlLong, jlLong), Long, jlLong] {
def outputEncoder: Encoder[jlLong] = Encoders.LONG
}
-final case class Reduce[T: Encoder](r: (T, T) => T)(implicit i: Encoder[Option[T]])
- extends Aggregator[T, Option[T], T] {
- def zero: Option[T] = None
- def reduce(b: Option[T], a: T): Option[T] = Some(b.fold(a)(r(_, a)))
- def merge(b1: Option[T], b2: Option[T]): Option[T] =
- (b1, b2) match {
- case (Some(a), Some(b)) => Some(r(a, b))
- case (Some(a), None) => Some(a)
- case (None, Some(b)) => Some(b)
- case (None, None) => None
- }
- def finish(reduction: Option[T]): T = reduction.get
- def bufferEncoder: Encoder[Option[T]] = implicitly
- def outputEncoder: Encoder[T] = implicitly
-}
-
@SQLUserDefinedType(udt = classOf[CountSerDeUDT])
case class CountSerDeSQL(nSer: Int, nDeSer: Int, sum: Int)
@@ -196,9 +180,6 @@ abstract class UDAQuerySuite extends QueryTest with SQLTestUtils with TestHiveSi
val data4 = Seq[Boolean](true, false, true).toDF("boolvalues")
data4.write.saveAsTable("agg4")
- val data5 = Seq[(Int, (Int, Int))]((1, (2, 3))).toDF("key", "value")
- data5.write.saveAsTable("agg5")
-
val emptyDF = spark.createDataFrame(
sparkContext.emptyRDD[Row],
StructType(StructField("key", StringType) :: StructField("value", IntegerType) :: Nil))
@@ -209,8 +190,6 @@ abstract class UDAQuerySuite extends QueryTest with SQLTestUtils with TestHiveSi
spark.udf.register("mydoubleavg", udaf(MyDoubleAvgAgg))
spark.udf.register("longProductSum", udaf(LongProductSumAgg))
spark.udf.register("arraysum", udaf(ArrayDataAgg))
- spark.udf.register("reduceOptionPair", udaf(Reduce[Option[(Int, Int)]](
- (opt1, opt2) => opt1.zip(opt2).map { case ((a1, b1), (a2, b2)) => (a1 + a2, b1 + b2) })))
}
override def afterAll(): Unit = {
@@ -392,12 +371,6 @@ abstract class UDAQuerySuite extends QueryTest with SQLTestUtils with TestHiveSi
Row(Seq(12.0, 15.0, 18.0)) :: Nil)
}
- test("SPARK-52023: Returning Option[Product] from udaf") {
- checkAnswer(
- spark.sql("SELECT reduceOptionPair(value) FROM agg5 GROUP BY key"),
- Row(Row(2, 3)) :: Nil)
- }
-
test("verify aggregator ser/de behavior") {
val data = sparkContext.parallelize((1 to 100).toSeq, 3).toDF("value1")
val agg = udaf(CountSerDeAgg)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org