[SPARK-58043][SQL] Use explicit .apply instead of a case class companion as a function for Scala 3 compatibility

python/pyspark/sql/tests/streaming/test_streaming_listener.py

@@ -288,7 +288,9 @@ def get_number_of_public_methods(clz):
             get_number_of_public_methods(
                 "org.apache.spark.sql.streaming.StreamingQueryListener$QueryStartedEvent"
             ),
-            16,
+            # 17 (not 16): jobTags serialization passes JString.apply, whose eta-expansion
+            # adds a synthetic public $anonfun method on this class in the Scala 2.13 build.
+            17,
             msg,
         )
         self.assertEqual(

sql/api/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryListener.scala

@@ -181,7 +181,7 @@ object StreamingQueryListener extends Serializable {
         ("runId" -> JString(runId.toString)) ~
         ("name" -> JString(name)) ~
         ("timestamp" -> JString(timestamp)) ~
-        ("jobTags" -> JArray(jobTags.toList.map(JString)))
+        ("jobTags" -> JArray(jobTags.toList.map(JString.apply)))
     }
   }
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/StructFilters.scala

@@ -63,7 +63,7 @@ abstract class StructFilters(pushedFilters: Seq[sources.Filter], schema: StructT
     val reducedExpr = filters
       .sortBy(_.references.length)
       .flatMap(filterToExpression(_, toRef))
-      .reduce(And)
+      .reduce(And.apply)
     Predicate.create(reducedExpr)
   }
 
@@ -122,15 +122,15 @@ object StructFilters {
       case sources.Or(left, right) =>
         zip(translate(left), translate(right)).map(Or.tupled)
       case sources.Not(child) =>
-        translate(child).map(Not)
+        translate(child).map(Not.apply)
       case sources.EqualTo(attribute, value) =>
         zipAttributeAndValue(attribute, value).map(EqualTo.tupled)
       case sources.EqualNullSafe(attribute, value) =>
         zipAttributeAndValue(attribute, value).map(EqualNullSafe.tupled)
       case sources.IsNull(attribute) =>
-        toRef(attribute).map(IsNull)
+        toRef(attribute).map(IsNull.apply)
       case sources.IsNotNull(attribute) =>
-        toRef(attribute).map(IsNotNull)
+        toRef(attribute).map(IsNotNull.apply)
       case sources.In(attribute, values) =>
         val literals = values.toImmutableArraySeq.flatMap(toLiteral)
         if (literals.length == values.length) {

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

@@ -1519,7 +1519,7 @@ class Analyzer(
             case None =>
               throw QueryCompilationErrors.missingStaticPartitionColumn(name)
           }
-        }.reduce(And)
+        }.reduce(And.apply)
       }
     }
   }
@@ -3777,7 +3777,7 @@ class Analyzer(
           val inputNullCheck = inputPrimitivesPair.collect {
             case (isPrimitive, input) if isPrimitive && input.nullable =>
               IsNull(input)
-          }.reduceLeftOption[Expression](Or)
+          }.reduceLeftOption[Expression](Or.apply)
 
           if (inputNullCheck.isDefined) {
             // Once we add an `If` check above the udf, it is safe to mark those checked inputs

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/NaturalAndUsingJoinResolution.scala

@@ -70,7 +70,7 @@ object NaturalAndUsingJoinResolution extends DataTypeErrorsBase with SQLConfHelp
     )
     val joinPairs = leftKeys.zip(rightKeys)
 
-    val newCondition = (condition ++ joinPairs.map(EqualTo.tupled)).reduceOption(And)
+    val newCondition = (condition ++ joinPairs.map(EqualTo.tupled)).reduceOption(And.apply)
 
     // the output list looks like: join keys, columns from left, columns from right
     val (output, hiddenOutput) = computeOutputAndHiddenOutput(

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveTableConstraints.scala

@@ -57,7 +57,7 @@ class ResolveTableConstraints(val catalogManager: CatalogManager) extends Rule[L
           val genColInvariants = buildGeneratedColumnConstraints(r, v2Write)
           val allInvariants = tableCheckInvariants ++ genColInvariants
           // Combine the check invariants into a single expression using conjunctive AND.
-          allInvariants.reduceOption(And).fold(v2Write)(
+          allInvariants.reduceOption(And.apply).fold(v2Write)(
             condition => v2Write.withNewQuery(Filter(condition, v2Write.query)))
         case _ =>
           v2Write

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala

@@ -392,8 +392,8 @@ object RewriteMergeIntoTable extends RewriteRowLevelCommand with PredicateHelper
     val (targetPredicates, joinPredicates) = predicates.partition { predicate =>
       predicate.references.subsetOf(targetTable.outputSet)
     }
-    val targetCond = targetPredicates.reduceOption(And).getOrElse(TrueLiteral)
-    val joinCond = joinPredicates.reduceOption(And).getOrElse(TrueLiteral)
+    val targetCond = targetPredicates.reduceOption(And.apply).getOrElse(TrueLiteral)
+    val joinCond = joinPredicates.reduceOption(And.apply).getOrElse(TrueLiteral)
     (Filter(targetCond, targetTable), joinCond)
   }
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/ExternalCatalogUtils.scala

@@ -205,7 +205,7 @@ object ExternalCatalogUtils {
         nonPartitionPruningPredicates)
     }
 
-    Predicate.createInterpreted(predicates.reduce(And).transform {
+    Predicate.createInterpreted(predicates.reduce(And.apply).transform {
       case att: AttributeReference =>
         val index = partitionSchema.indexWhere(_.name == att.name)
         BoundReference(index, partitionSchema(index).dataType, nullable = true)

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtils.scala

@@ -245,21 +245,21 @@ object V2ExpressionUtils extends SQLConfHelper with Logging {
 
   private def convertPredicate(expr: GeneralScalarExpression): Option[Expression] = {
     expr.name match {
-      case "IS_NULL" => convertUnaryExpr(expr, IsNull)
-      case "IS_NOT_NULL" => convertUnaryExpr(expr, IsNotNull)
-      case "NOT" => convertUnaryExpr(expr, Not)
-      case "=" => convertBinaryExpr(expr, EqualTo)
-      case "<=>" => convertBinaryExpr(expr, EqualNullSafe)
-      case ">" => convertBinaryExpr(expr, GreaterThan)
-      case ">=" => convertBinaryExpr(expr, GreaterThanOrEqual)
-      case "<" => convertBinaryExpr(expr, LessThan)
-      case "<=" => convertBinaryExpr(expr, LessThanOrEqual)
+      case "IS_NULL" => convertUnaryExpr(expr, IsNull.apply)
+      case "IS_NOT_NULL" => convertUnaryExpr(expr, IsNotNull.apply)
+      case "NOT" => convertUnaryExpr(expr, Not.apply)
+      case "=" => convertBinaryExpr(expr, EqualTo.apply)
+      case "<=>" => convertBinaryExpr(expr, EqualNullSafe.apply)
+      case ">" => convertBinaryExpr(expr, GreaterThan.apply)
+      case ">=" => convertBinaryExpr(expr, GreaterThanOrEqual.apply)
+      case "<" => convertBinaryExpr(expr, LessThan.apply)
+      case "<=" => convertBinaryExpr(expr, LessThanOrEqual.apply)
       case "<>" => convertBinaryExpr(expr, (left, right) => Not(EqualTo(left, right)))
-      case "AND" => convertBinaryExpr(expr, And)
-      case "OR" => convertBinaryExpr(expr, Or)
-      case "STARTS_WITH" => convertBinaryExpr(expr, StartsWith)
-      case "ENDS_WITH" => convertBinaryExpr(expr, EndsWith)
-      case "CONTAINS" => convertBinaryExpr(expr, Contains)
+      case "AND" => convertBinaryExpr(expr, And.apply)
+      case "OR" => convertBinaryExpr(expr, Or.apply)
+      case "STARTS_WITH" => convertBinaryExpr(expr, StartsWith.apply)
+      case "ENDS_WITH" => convertBinaryExpr(expr, EndsWith.apply)
+      case "CONTAINS" => convertBinaryExpr(expr, Contains.apply)
       case "IN" => convertExpr(expr, children => In(children.head, children.tail))
       case "BOOLEAN_EXPRESSION" => toCatalyst(expr.children().head)
       case _ => None
@@ -297,9 +297,9 @@ object V2ExpressionUtils extends SQLConfHelper with Logging {
       case "/" => convertBinaryExpr(expr, Divide(_, _, evalMode = EvalMode.ANSI))
       case "%" => convertBinaryExpr(expr, Remainder(_, _, evalMode = EvalMode.ANSI))
       case "ABS" => convertUnaryExpr(expr, Abs(_, failOnError = true))
-      case "COALESCE" => convertExpr(expr, Coalesce)
-      case "GREATEST" => convertExpr(expr, Greatest)
-      case "LEAST" => convertExpr(expr, Least)
+      case "COALESCE" => convertExpr(expr, Coalesce.apply)
+      case "GREATEST" => convertExpr(expr, Greatest.apply)
+      case "LEAST" => convertExpr(expr, Least.apply)
       case "RAND" =>
         if (expr.children.isEmpty) {
           Some(new Rand())
@@ -308,20 +308,20 @@ object V2ExpressionUtils extends SQLConfHelper with Logging {
         } else {
           None
         }
-      case "LOG" => convertBinaryExpr(expr, Logarithm)
-      case "LOG10" => convertUnaryExpr(expr, Log10)
-      case "LOG2" => convertUnaryExpr(expr, Log2)
-      case "LN" => convertUnaryExpr(expr, Log)
-      case "EXP" => convertUnaryExpr(expr, Exp)
-      case "POWER" => convertBinaryExpr(expr, Pow)
-      case "SQRT" => convertUnaryExpr(expr, Sqrt)
+      case "LOG" => convertBinaryExpr(expr, Logarithm.apply)
+      case "LOG10" => convertUnaryExpr(expr, Log10.apply)
+      case "LOG2" => convertUnaryExpr(expr, Log2.apply)
+      case "LN" => convertUnaryExpr(expr, Log.apply)
+      case "EXP" => convertUnaryExpr(expr, Exp.apply)
+      case "POWER" => convertBinaryExpr(expr, Pow.apply)
+      case "SQRT" => convertUnaryExpr(expr, Sqrt.apply)
       case "FLOOR" => convertUnaryExpr(expr, Floor(_))
       case "CEIL" => convertUnaryExpr(expr, Ceil(_))
       case "ROUND" => convertBinaryExpr(expr, Round(_, _, ansiEnabled = true))
-      case "CBRT" => convertUnaryExpr(expr, Cbrt)
-      case "DEGREES" => convertUnaryExpr(expr, ToDegrees)
-      case "RADIANS" => convertUnaryExpr(expr, ToRadians)
-      case "SIGN" => convertUnaryExpr(expr, Signum)
+      case "CBRT" => convertUnaryExpr(expr, Cbrt.apply)
+      case "DEGREES" => convertUnaryExpr(expr, ToDegrees.apply)
+      case "RADIANS" => convertUnaryExpr(expr, ToRadians.apply)
+      case "SIGN" => convertUnaryExpr(expr, Signum.apply)
       case "WIDTH_BUCKET" =>
         convertExpr(
           expr,
@@ -332,30 +332,30 @@ object V2ExpressionUtils extends SQLConfHelper with Logging {
 
   private def convertTrigonometricFunc(expr: GeneralScalarExpression): Option[Expression] = {
     expr.name match {
-      case "SIN" => convertUnaryExpr(expr, Sin)
-      case "SINH" => convertUnaryExpr(expr, Sinh)
-      case "COS" => convertUnaryExpr(expr, Cos)
-      case "COSH" => convertUnaryExpr(expr, Cosh)
-      case "TAN" => convertUnaryExpr(expr, Tan)
-      case "TANH" => convertUnaryExpr(expr, Tanh)
-      case "COT" => convertUnaryExpr(expr, Cot)
-      case "ASIN" => convertUnaryExpr(expr, Asin)
-      case "ASINH" => convertUnaryExpr(expr, Asinh)
-      case "ACOS" => convertUnaryExpr(expr, Acos)
-      case "ACOSH" => convertUnaryExpr(expr, Acosh)
-      case "ATAN" => convertUnaryExpr(expr, Atan)
-      case "ATANH" => convertUnaryExpr(expr, Atanh)
-      case "ATAN2" => convertBinaryExpr(expr, Atan2)
+      case "SIN" => convertUnaryExpr(expr, Sin.apply)
+      case "SINH" => convertUnaryExpr(expr, Sinh.apply)
+      case "COS" => convertUnaryExpr(expr, Cos.apply)
+      case "COSH" => convertUnaryExpr(expr, Cosh.apply)
+      case "TAN" => convertUnaryExpr(expr, Tan.apply)
+      case "TANH" => convertUnaryExpr(expr, Tanh.apply)
+      case "COT" => convertUnaryExpr(expr, Cot.apply)
+      case "ASIN" => convertUnaryExpr(expr, Asin.apply)
+      case "ASINH" => convertUnaryExpr(expr, Asinh.apply)
+      case "ACOS" => convertUnaryExpr(expr, Acos.apply)
+      case "ACOSH" => convertUnaryExpr(expr, Acosh.apply)
+      case "ATAN" => convertUnaryExpr(expr, Atan.apply)
+      case "ATANH" => convertUnaryExpr(expr, Atanh.apply)
+      case "ATAN2" => convertBinaryExpr(expr, Atan2.apply)
       case _ => None
     }
   }
 
   private def convertBitwiseFunc(expr: GeneralScalarExpression): Option[Expression] = {
     expr.name match {
-      case "~" => convertUnaryExpr(expr, BitwiseNot)
-      case "&" => convertBinaryExpr(expr, BitwiseAnd)
-      case "|" => convertBinaryExpr(expr, BitwiseOr)
-      case "^" => convertBinaryExpr(expr, BitwiseXor)
+      case "~" => convertUnaryExpr(expr, BitwiseNot.apply)
+      case "&" => convertBinaryExpr(expr, BitwiseAnd.apply)
+      case "|" => convertBinaryExpr(expr, BitwiseOr.apply)
+      case "^" => convertBinaryExpr(expr, BitwiseXor.apply)
       case _ => None
     }
   }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Count.scala

@@ -94,7 +94,7 @@ case class Count(children: Seq[Expression]) extends DeclarativeAggregate
       )
     } else {
       Seq(
-        /* count = */ If(nullableChildren.map(IsNull).reduce(Or), count, count + 1L)
+        /* count = */ If(nullableChildren.map(IsNull.apply).reduce(Or.apply), count, count + 1L)
       )
     }
   }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/json/JsonExpressionEvalUtils.scala

@@ -62,7 +62,7 @@ object JsonPathParser extends RegexParsers {
   // parse `[*]` and `[123]` subscripts
   def subscript: Parser[List[PathInstruction]] =
     for {
-      operand <- '[' ~> ('*' ^^^ Wildcard | long ^^ Index) <~ ']'
+      operand <- '[' ~> ('*' ^^^ Wildcard | long ^^ Index.apply) <~ ']'
     } yield {
       Subscript :: operand :: Nil
     }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/windowExpressions.scala

@@ -971,7 +971,7 @@ abstract class RankLike extends AggregateWindowFunction {
   /** Predicate that detects if the order attributes have changed. */
   protected val orderEquals = children.zip(orderAttrs)
     .map(EqualNullSafe.tupled)
-    .reduceOption(And)
+    .reduceOption(And.apply)
     .getOrElse(Literal(true))
 
   protected val orderInit = children.map(e => Literal.create(null, e.dataType))

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/CostBasedJoinReorder.scala

@@ -304,7 +304,7 @@ object JoinReorderDP extends PredicateHelper with Logging {
     } else {
       (otherPlan, onePlan)
     }
-    val newJoin = Join(left, right, Inner, joinConds.reduceOption(And), JoinHint.NONE)
+    val newJoin = Join(left, right, Inner, joinConds.reduceOption(And.apply), JoinHint.NONE)
     val collectedJoinConds = joinConds ++ oneJoinPlan.joinConds ++ otherJoinPlan.joinConds
     val remainingConds = conditions -- collectedJoinConds
     val neededAttr = AttributeSet(remainingConds.flatMap(_.references)) ++ topOutput

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/DecorrelateInnerQuery.scala

@@ -623,7 +623,7 @@ object DecorrelateInnerQuery extends PredicateHelper {
               val newFilterCond = newCorrelated ++ uncorrelated
               val newFilter = newFilterCond match {
                 case Nil => newChild
-                case conditions => Filter(conditions.reduce(And), newChild)
+                case conditions => Filter(conditions.reduce(And.apply), newChild)
               }
               // Equality predicates are used as join conditions with the outer query.
               val newJoinCond = joinCond ++ equalityCond
@@ -638,7 +638,7 @@ object DecorrelateInnerQuery extends PredicateHelper {
               val newOuterReferenceMap = outerReferenceMap ++ equivalences
               val newFilter = uncorrelated match {
                 case Nil => newChild
-                case conditions => Filter(conditions.reduce(And), newChild)
+                case conditions => Filter(conditions.reduce(And.apply), newChild)
               }
               val newJoinCond = joinCond ++ correlated
               (newFilter, newJoinCond, newOuterReferenceMap)
@@ -907,7 +907,7 @@ object DecorrelateInnerQuery extends PredicateHelper {
               // Use the current join conditions returned from the recursive call as the join
               // conditions for the left outer join. All outer references in the join
               // conditions are replaced by the newly created domain attributes.
-              val condition = replaceOuterReferences(joinCond, mapping).reduceOption(And)
+              val condition = replaceOuterReferences(joinCond, mapping).reduceOption(And.apply)
               val domainJoin = DomainJoin(domainAttrs, agg, LeftOuter, condition)
               // Original domain attributes preserved through Aggregate are no longer needed.
               val newProjectList = projectList.filter(!referencesToAdd.contains(_))
@@ -1034,7 +1034,7 @@ object DecorrelateInnerQuery extends PredicateHelper {
               case (outer, inner) => rightOuterReferenceMap.get(outer).map(EqualNullSafe(inner, _))
             }
             val newCondition = (newCorrelated ++ uncorrelated
-              ++ augmentedConditions).reduceOption(And)
+              ++ augmentedConditions).reduceOption(And.apply)
             val newJoin = j.copy(left = newLeft, right = newRight, condition = newCondition)
             (newJoin, newJoinCond, newOuterReferenceMap)
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/NormalizeFloatingNumbers.scala

@@ -92,7 +92,7 @@ object NormalizeFloatingNumbers extends Rule[LogicalPlan] {
           val newConditions = newLeftJoinKeys.zip(newRightJoinKeys).map {
             case (l, r) => EqualTo(l, r)
           } ++ condition
-          j.copy(condition = Some(newConditions.reduce(And)))
+          j.copy(condition = Some(newConditions.reduce(And.apply)))
 
         // TODO: ideally Aggregate should also be handled here, but its grouping expressions are
         // mixed in its aggregate expressions. It's unreliable to change the grouping expressions