diff --git a/python/pyspark/sql/tests/streaming/test_streaming_listener.py b/python/pyspark/sql/tests/streaming/test_streaming_listener.py index b4922f54b2170..5b575b46abbe1 100644 --- a/python/pyspark/sql/tests/streaming/test_streaming_listener.py +++ b/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( diff --git a/sql/api/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryListener.scala b/sql/api/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryListener.scala index 88b39ed5d8167..501202fb9cb90 100644 --- a/sql/api/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryListener.scala +++ b/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))) } } diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/StructFilters.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/StructFilters.scala index 1b2013d87eedf..f78470557e698 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/StructFilters.scala +++ b/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) { diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala index 49fdbd45b91c2..8743a7f097bee 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala +++ b/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 diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/NaturalAndUsingJoinResolution.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/NaturalAndUsingJoinResolution.scala index 4eba9b5f8afaa..ae86f05c93444 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/NaturalAndUsingJoinResolution.scala +++ b/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( diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveTableConstraints.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveTableConstraints.scala index de4490143a83d..49632e9e1c6a6 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveTableConstraints.scala +++ b/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 diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala index e11a985d85ddb..f652bdc2365b6 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala +++ b/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) } diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/ExternalCatalogUtils.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/ExternalCatalogUtils.scala index 8960c7345521c..5779174b58030 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/ExternalCatalogUtils.scala +++ b/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) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtils.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtils.scala index 702255e075743..01d7826c9505e 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtils.scala +++ b/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 } } diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Count.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Count.scala index 758ef22f0a2c2..97cf01ad95d57 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Count.scala +++ b/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) ) } } diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/json/JsonExpressionEvalUtils.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/json/JsonExpressionEvalUtils.scala index e80b4a355b726..63c568b2a66e1 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/json/JsonExpressionEvalUtils.scala +++ b/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 } diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/windowExpressions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/windowExpressions.scala index f26a4dc9a0c0c..4261b3dd39abe 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/windowExpressions.scala +++ b/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)) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/CostBasedJoinReorder.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/CostBasedJoinReorder.scala index 471f0bd554105..1795495991928 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/CostBasedJoinReorder.scala +++ b/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 diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/DecorrelateInnerQuery.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/DecorrelateInnerQuery.scala index c6a7940b93aaf..be2a0db3b4f36 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/DecorrelateInnerQuery.scala +++ b/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) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/NormalizeFloatingNumbers.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/NormalizeFloatingNumbers.scala index 1d25788fdb6c2..00ea6f56ee2e3 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/NormalizeFloatingNumbers.scala +++ b/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 diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala index 597a70e01a6e9..3a17392e5cb5e 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala @@ -442,8 +442,11 @@ abstract class Optimizer(catalogManager: CatalogManager) } } val newPlan = Project(projections, - if (predicates.nonEmpty) Filter(predicates.reduce(And), optimizedAgg) else optimizedAgg - ) + if (predicates.nonEmpty) { + Filter(predicates.reduce(And.apply), optimizedAgg) + } else { + optimizedAgg + }) val needTopLevelProject = newPlan.output.zip(optimizedAgg.output).exists { case (oldAttr, newAttr) => oldAttr.exprId != newAttr.exprId } @@ -1755,7 +1758,7 @@ object InferFiltersFromGenerate extends Rule[LogicalPlan] { ) -- generate.child.constraints if (inferredFilters.nonEmpty) { - generate.copy(child = Filter(inferredFilters.reduce(And), generate.child)) + generate.copy(child = Filter(inferredFilters.reduce(And.apply), generate.child)) } else { generate } @@ -1797,7 +1800,7 @@ object InferFiltersFromConstraints extends Rule[LogicalPlan] val newFilters = filter.constraints -- (child.constraints ++ splitConjunctivePredicates(condition)) if (newFilters.nonEmpty) { - Filter(And(newFilters.reduce(And), condition), child) + Filter(And(newFilters.reduce(And.apply), condition), child) } else { filter } @@ -1848,7 +1851,7 @@ object InferFiltersFromConstraints extends Rule[LogicalPlan] if (newPredicates.isEmpty) { plan } else { - Filter(newPredicates.reduce(And), plan) + Filter(newPredicates.reduce(And.apply), plan) } } } @@ -1948,13 +1951,13 @@ object CombineFilters extends Rule[LogicalPlan] with PredicateHelper { val (combineCandidates, rest) = splitConjunctivePredicates(fc).partition(p => p.deterministic && !p.throwable) val mergedFilter = (ExpressionSet(combineCandidates) -- - ExpressionSet(splitConjunctivePredicates(nc))).reduceOption(And) match { + ExpressionSet(splitConjunctivePredicates(nc))).reduceOption(And.apply) match { case Some(ac) => Filter(And(nc, ac), grandChild) case None => nf } - rest.reduceOption(And).map(c => Filter(c, mergedFilter)).getOrElse(mergedFilter) + rest.reduceOption(And.apply).map(c => Filter(c, mergedFilter)).getOrElse(mergedFilter) } } @@ -2078,7 +2081,7 @@ object PruneFilters extends Rule[LogicalPlan] with PredicateHelper { } else if (remainingPredicates.isEmpty) { p } else { - val newCond = remainingPredicates.reduce(And) + val newCond = remainingPredicates.reduce(And.apply) Filter(newCond, p) } } @@ -2173,7 +2176,7 @@ object PushPredicateThroughNonJoin extends Rule[LogicalPlan] with PredicateHelpe // Make a base instance which has all of the cheap filters pushed down. // For all filter which do not reference any expensive aliases then // just push the filter while resolving the non-expensive aliases. - val combinedCheapFilter = cheap.reduce(And) + val combinedCheapFilter = cheap.reduce(And.apply) val baseChild = Filter(combinedCheapFilter, child = grandChild) // Take our projection and place it on top of the pushed filters. val topProjection = project.copy(child = baseChild) @@ -2184,7 +2187,7 @@ object PushPredicateThroughNonJoin extends Rule[LogicalPlan] with PredicateHelpe } else { // Finally add any filters which could not be pushed val remainingConditions = expensiveWithUsed.map(_._1) - Filter(remainingConditions.reduce(And), topProjection) + Filter(remainingConditions.reduce(And.apply), topProjection) } } } @@ -2207,12 +2210,12 @@ object PushPredicateThroughNonJoin extends Rule[LogicalPlan] with PredicateHelpe } if (pushDown.nonEmpty) { - val pushDownPredicate = pushDown.reduce(And) + val pushDownPredicate = pushDown.reduce(And.apply) val replaced = replaceAlias(pushDownPredicate, aliasMap) val newAggregate = aggregate.copy(child = Filter(replaced, aggregate.child)) // If there is no more filter to stay up, just eliminate the filter. // Otherwise, create "Filter(stayUp) <- Aggregate <- Filter(pushDownPredicate)". - if (stayUp.isEmpty) newAggregate else Filter(stayUp.reduce(And), newAggregate) + if (stayUp.isEmpty) newAggregate else Filter(stayUp.reduce(And.apply), newAggregate) } else { filter } @@ -2236,9 +2239,9 @@ object PushPredicateThroughNonJoin extends Rule[LogicalPlan] with PredicateHelpe val stayUp = rest ++ nonDeterministic if (pushDown.nonEmpty) { - val pushDownPredicate = pushDown.reduce(And) + val pushDownPredicate = pushDown.reduce(And.apply) val newWindow = w.copy(child = Filter(pushDownPredicate, w.child)) - if (stayUp.isEmpty) newWindow else Filter(stayUp.reduce(And), newWindow) + if (stayUp.isEmpty) newWindow else Filter(stayUp.reduce(And.apply), newWindow) } else { filter } @@ -2248,7 +2251,7 @@ object PushPredicateThroughNonJoin extends Rule[LogicalPlan] with PredicateHelpe val (pushDown, stayUp) = splitConjunctivePredicates(condition).partition(_.deterministic) if (pushDown.nonEmpty) { - val pushDownCond = pushDown.reduceLeft(And) + val pushDownCond = pushDown.reduceLeft(And.apply) // The union is the child of the filter so it's children are grandchildren. // Moves filters down to the grandchild if there is an element in the grand child's // output which is semantically equal to the filter being evaluated. @@ -2264,7 +2267,7 @@ object PushPredicateThroughNonJoin extends Rule[LogicalPlan] with PredicateHelpe val newUnion = union.withNewChildren(newGrandChildren) if (stayUp.nonEmpty) { // If there is any filter we can't push evaluate them post union - Filter(stayUp.reduceLeft(And), newUnion) + Filter(stayUp.reduceLeft(And.apply), newUnion) } else { // If we pushed all filters then just return the new union. newUnion @@ -2280,11 +2283,11 @@ object PushPredicateThroughNonJoin extends Rule[LogicalPlan] with PredicateHelpe } if (pushDown.nonEmpty) { - val pushDownPredicate = pushDown.reduceLeft(And) + val pushDownPredicate = pushDown.reduceLeft(And.apply) val newWatermark = watermark.copy(child = Filter(pushDownPredicate, watermark.child)) // If there is no more filter to stay up, just eliminate the filter. // Otherwise, create "Filter(stayUp) <- watermark <- Filter(pushDownPredicate)". - if (stayUp.isEmpty) newWatermark else Filter(stayUp.reduceLeft(And), newWatermark) + if (stayUp.isEmpty) newWatermark else Filter(stayUp.reduceLeft(And.apply), newWatermark) } else { filter } @@ -2331,9 +2334,9 @@ object PushPredicateThroughNonJoin extends Rule[LogicalPlan] with PredicateHelpe val stayUp = rest ++ nonDeterministic if (pushDown.nonEmpty) { - val newChild = insertFilter(pushDown.reduceLeft(And)) + val newChild = insertFilter(pushDown.reduceLeft(And.apply)) if (stayUp.nonEmpty) { - Filter(stayUp.reduceLeft(And), newChild) + Filter(stayUp.reduceLeft(And.apply), newChild) } else { newChild } @@ -2404,17 +2407,17 @@ object PushPredicateThroughJoin extends Rule[LogicalPlan] with PredicateHelper { case _: InnerLike => // push down the single side `where` condition into respective sides val newLeft = leftFilterConditions. - reduceLeftOption(And).map(Filter(_, left)).getOrElse(left) + reduceLeftOption(And.apply).map(Filter(_, left)).getOrElse(left) val newRight = rightFilterConditions. - reduceLeftOption(And).map(Filter(_, right)).getOrElse(right) + reduceLeftOption(And.apply).map(Filter(_, right)).getOrElse(right) // don't push throwable expressions into join condition val (newJoinConditions, others) = commonFilterCondition.partition(cond => canEvaluateWithinJoin(cond) && !cond.throwable) - val newJoinCond = (newJoinConditions ++ joinCondition).reduceLeftOption(And) + val newJoinCond = (newJoinConditions ++ joinCondition).reduceLeftOption(And.apply) val join = Join(newLeft, newRight, joinType, newJoinCond, hint) if (others.nonEmpty) { - Filter(others.reduceLeft(And), join) + Filter(others.reduceLeft(And.apply), join) } else { join } @@ -2422,22 +2425,22 @@ object PushPredicateThroughJoin extends Rule[LogicalPlan] with PredicateHelper { // push down the right side only `where` condition val newLeft = left val newRight = rightFilterConditions. - reduceLeftOption(And).map(Filter(_, right)).getOrElse(right) + reduceLeftOption(And.apply).map(Filter(_, right)).getOrElse(right) val newJoinCond = joinCondition val newJoin = Join(newLeft, newRight, RightOuter, newJoinCond, hint) (leftFilterConditions ++ commonFilterCondition). - reduceLeftOption(And).map(Filter(_, newJoin)).getOrElse(newJoin) + reduceLeftOption(And.apply).map(Filter(_, newJoin)).getOrElse(newJoin) case LeftOuter | LeftSingle | LeftExistence(_) => // push down the left side only `where` condition val newLeft = leftFilterConditions. - reduceLeftOption(And).map(Filter(_, left)).getOrElse(left) + reduceLeftOption(And.apply).map(Filter(_, left)).getOrElse(left) val newRight = right val newJoinCond = joinCondition val newJoin = Join(newLeft, newRight, joinType, newJoinCond, hint) (rightFilterConditions ++ commonFilterCondition). - reduceLeftOption(And).map(Filter(_, newJoin)).getOrElse(newJoin) + reduceLeftOption(And.apply).map(Filter(_, newJoin)).getOrElse(newJoin) case other => throw SparkException.internalError(s"Unexpected join type: $other") @@ -2452,26 +2455,26 @@ object PushPredicateThroughJoin extends Rule[LogicalPlan] with PredicateHelper { case _: InnerLike | LeftSemi => // push down the single side only join filter for both sides sub queries val newLeft = leftJoinConditions. - reduceLeftOption(And).map(Filter(_, left)).getOrElse(left) + reduceLeftOption(And.apply).map(Filter(_, left)).getOrElse(left) val newRight = rightJoinConditions. - reduceLeftOption(And).map(Filter(_, right)).getOrElse(right) - val newJoinCond = commonJoinCondition.reduceLeftOption(And) + reduceLeftOption(And.apply).map(Filter(_, right)).getOrElse(right) + val newJoinCond = commonJoinCondition.reduceLeftOption(And.apply) Join(newLeft, newRight, joinType, newJoinCond, hint) case RightOuter => // push down the left side only join filter for left side sub query val newLeft = leftJoinConditions. - reduceLeftOption(And).map(Filter(_, left)).getOrElse(left) + reduceLeftOption(And.apply).map(Filter(_, left)).getOrElse(left) val newRight = right - val newJoinCond = (rightJoinConditions ++ commonJoinCondition).reduceLeftOption(And) + val newJoinCond = (rightJoinConditions ++ commonJoinCondition).reduceLeftOption(And.apply) Join(newLeft, newRight, RightOuter, newJoinCond, hint) case LeftOuter | LeftAnti | ExistenceJoin(_) => // push down the right side only join filter for right sub query val newLeft = left val newRight = rightJoinConditions. - reduceLeftOption(And).map(Filter(_, right)).getOrElse(right) - val newJoinCond = (leftJoinConditions ++ commonJoinCondition).reduceLeftOption(And) + reduceLeftOption(And.apply).map(Filter(_, right)).getOrElse(right) + val newJoinCond = (leftJoinConditions ++ commonJoinCondition).reduceLeftOption(And.apply) Join(newLeft, newRight, joinType, newJoinCond, hint) // Do not move join predicates of a single join. @@ -2778,7 +2781,7 @@ object ReplaceIntersectWithSemiJoin extends Rule[LogicalPlan] { case Intersect(left, right, false) => assert(left.output.size == right.output.size) val joinCond = left.output.zip(right.output).map { case (l, r) => EqualNullSafe(l, r) } - Distinct(Join(left, right, LeftSemi, joinCond.reduceLeftOption(And), JoinHint.NONE)) + Distinct(Join(left, right, LeftSemi, joinCond.reduceLeftOption(And.apply), JoinHint.NONE)) } } @@ -2800,7 +2803,7 @@ object ReplaceExceptWithAntiJoin extends Rule[LogicalPlan] { case Except(left, right, false) => assert(left.output.size == right.output.size) val joinCond = left.output.zip(right.output).map { case (l, r) => EqualNullSafe(l, r) } - Distinct(Join(left, right, LeftAnti, joinCond.reduceLeftOption(And), JoinHint.NONE)) + Distinct(Join(left, right, LeftAnti, joinCond.reduceLeftOption(And.apply), JoinHint.NONE)) } } diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushDownLeftSemiAntiJoin.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushDownLeftSemiAntiJoin.scala index a5fcbe6f16b38..7aec47054f75a 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushDownLeftSemiAntiJoin.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushDownLeftSemiAntiJoin.scala @@ -73,7 +73,7 @@ object PushDownLeftSemiAntiJoin extends Rule[LogicalPlan] replaced.references.subsetOf(agg.child.outputSet ++ rightOp.outputSet) } val makeJoinCondition = (predicates: Seq[Expression]) => { - replaceAlias(predicates.reduce(And), aliasMap) + replaceAlias(predicates.reduce(And.apply), aliasMap) } pushDownJoin(join, canPushDownPredicate, makeJoinCondition) @@ -81,7 +81,7 @@ object PushDownLeftSemiAntiJoin extends Rule[LogicalPlan] case join @ Join(w: Window, rightOp, LeftSemiOrAnti(_), _, _) if w.partitionSpec.forall(_.isInstanceOf[AttributeReference]) => val partitionAttrs = AttributeSet(w.partitionSpec.flatMap(_.references)) ++ rightOp.outputSet - pushDownJoin(join, _.references.subsetOf(partitionAttrs), _.reduce(And)) + pushDownJoin(join, _.references.subsetOf(partitionAttrs), _.reduce(And.apply)) // LeftSemi/LeftAnti over Union case Join(union: Union, rightOp, LeftSemiOrAnti(joinType), joinCond, hint) @@ -107,7 +107,7 @@ object PushDownLeftSemiAntiJoin extends Rule[LogicalPlan] case join @ Join(u: UnaryNode, rightOp, LeftSemiOrAnti(_), _, _) if PushPredicateThroughNonJoin.canPushThrough(u) && u.expressions.forall(_.deterministic) => val validAttrs = u.child.outputSet ++ rightOp.outputSet - pushDownJoin(join, _.references.subsetOf(validAttrs), _.reduce(And)) + pushDownJoin(join, _.references.subsetOf(validAttrs), _.reduce(And.apply)) } /** @@ -160,7 +160,7 @@ object PushDownLeftSemiAntiJoin extends Rule[LogicalPlan] join.joinType match { // In case of Left semi join, the part of the join condition which does not refer to // to attributes of the grandchild are kept as a Filter above. - case LeftSemi => Filter(stayUp.reduce(And), newPlan) + case LeftSemi => Filter(stayUp.reduce(And.apply), newPlan) // In case of left-anti join, the join is pushed down only when the entire join // condition is eligible to be pushed down to preserve the semantics of left-anti join. case _ => join diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushExtraPredicateThroughJoin.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushExtraPredicateThroughJoin.scala index a2bc0bf83a2c6..93244075549b5 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushExtraPredicateThroughJoin.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushExtraPredicateThroughJoin.scala @@ -60,9 +60,9 @@ object PushExtraPredicateThroughJoin extends Rule[LogicalPlan] with PredicateHel j } else { lazy val newLeft = - leftExtraCondition.reduceLeftOption(And).map(Filter(_, left)).getOrElse(left) + leftExtraCondition.reduceLeftOption(And.apply).map(Filter(_, left)).getOrElse(left) lazy val newRight = - rightExtraCondition.reduceLeftOption(And).map(Filter(_, right)).getOrElse(right) + rightExtraCondition.reduceLeftOption(And.apply).map(Filter(_, right)).getOrElse(right) val newJoin = joinType match { case _: InnerLike | LeftSemi => diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushdownPredicatesAndPruneColumnsForCTEDef.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushdownPredicatesAndPruneColumnsForCTEDef.scala index 2339596920b01..06ae417327036 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushdownPredicatesAndPruneColumnsForCTEDef.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushdownPredicatesAndPruneColumnsForCTEDef.scala @@ -91,7 +91,7 @@ object PushdownPredicatesAndPruneColumnsForCTEDef extends Rule[LogicalPlan] { if (filteredPredicates.isEmpty) { Seq(Literal.TrueLiteral) } else { - preds :+ filteredPredicates.reduce(And) + preds :+ filteredPredicates.reduce(And.apply) } } val newAttributes = attrs ++ @@ -128,7 +128,7 @@ object PushdownPredicatesAndPruneColumnsForCTEDef extends Rule[LogicalPlan] { val preds = originalPlanWithPredicates.map(_._2).getOrElse(Seq.empty) if (!isTruePredicate(newPreds) && newPreds.exists(newPred => !preds.exists(_.semanticEquals(newPred)))) { - val newCombinedPred = newPreds.reduce(Or) + val newCombinedPred = newPreds.reduce(Or.apply) val newChild = if (needsPruning(originalPlan, newAttrSet)) { Project(newAttrSet.toSeq, originalPlan) } else { diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala index de847dddc9357..bd5bb62fa7cbc 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala @@ -209,7 +209,7 @@ object ConstantPropagation extends Rule[LogicalPlan] { case n: Not => // Ignore the EqualityPredicates from children since they are only propagated through And. val (newChild, _) = traverse(n.child, replaceChildren = true, nullIsFalse = false) - (newChild.map(Not), AttributeMap.empty) + (newChild.map(Not.apply), AttributeMap.empty) case _ => (None, AttributeMap.empty) } @@ -473,11 +473,11 @@ object BooleanSimplification extends Rule[LogicalPlan] with PredicateHelper { val rdiff = rhs.filterNot(e => common.exists(e.semanticEquals)) if (ldiff.isEmpty || rdiff.isEmpty) { // (a || b || c || ...) && (a || b) => (a || b) - common.reduce(Or) + common.reduce(Or.apply) } else { // (a || b || c || ...) && (a || b || d || ...) => // a || b || ((c || ...) && (d || ...)) - (common :+ And(ldiff.reduce(Or), rdiff.reduce(Or))).reduce(Or) + (common :+ And(ldiff.reduce(Or.apply), rdiff.reduce(Or.apply))).reduce(Or.apply) } } else { // No common factors from disjunctive predicates, reduce common factor from conjunction @@ -488,7 +488,7 @@ object BooleanSimplification extends Rule[LogicalPlan] with PredicateHelper { and } else { // (a && b) && a && (a && c) => a && b && c - buildBalancedPredicate(distinct.toSeq, And) + buildBalancedPredicate(distinct.toSeq, And.apply) } } @@ -511,11 +511,11 @@ object BooleanSimplification extends Rule[LogicalPlan] with PredicateHelper { val rdiff = rhs.filterNot(e => common.exists(e.semanticEquals)) if (ldiff.isEmpty || rdiff.isEmpty) { // (a && b) || (a && b && c && ...) => a && b - common.reduce(And) + common.reduce(And.apply) } else { // (a && b && c && ...) || (a && b && d && ...) => // a && b && ((c && ...) || (d && ...)) - (common :+ Or(ldiff.reduce(And), rdiff.reduce(And))).reduce(And) + (common :+ Or(ldiff.reduce(And.apply), rdiff.reduce(And.apply))).reduce(And.apply) } } else { // No common factors in conjunctive predicates, reduce common factor from disjunction @@ -526,7 +526,7 @@ object BooleanSimplification extends Rule[LogicalPlan] with PredicateHelper { or } else { // (a || b) || a || (a || c) => a || b || c - buildBalancedPredicate(distinct.toSeq, Or) + buildBalancedPredicate(distinct.toSeq, Or.apply) } } @@ -856,16 +856,16 @@ object LikeSimplification extends Rule[LogicalPlan] with PredicateHelper { } else { multi match { case l: LikeAll => - val and = buildBalancedPredicate(replacements, And) + val and = buildBalancedPredicate(replacements, And.apply) if (remainPatterns.nonEmpty) And(and, l.copy(patterns = remainPatterns)) else and case l: NotLikeAll => - val and = buildBalancedPredicate(replacements.map(Not(_)), And) + val and = buildBalancedPredicate(replacements.map(Not(_)), And.apply) if (remainPatterns.nonEmpty) And(and, l.copy(patterns = remainPatterns)) else and case l: LikeAny => - val or = buildBalancedPredicate(replacements, Or) + val or = buildBalancedPredicate(replacements, Or.apply) if (remainPatterns.nonEmpty) Or(or, l.copy(patterns = remainPatterns)) else or case l: NotLikeAny => - val or = buildBalancedPredicate(replacements.map(Not(_)), Or) + val or = buildBalancedPredicate(replacements.map(Not(_)), Or.apply) if (remainPatterns.nonEmpty) Or(or, l.copy(patterns = remainPatterns)) else or } } @@ -994,9 +994,9 @@ object NullDownPropagation extends Rule[LogicalPlan] { case q: LogicalPlan => q.transformExpressionsDownWithPruning( _.containsPattern(NULL_CHECK), ruleId) { case IsNull(e) if e.nullIntolerant && supportedNullIntolerant(e) => - e.children.map(IsNull(_): Expression).reduceLeft(Or) + e.children.map(IsNull(_): Expression).reduceLeft(Or.apply) case IsNotNull(e) if e.nullIntolerant && supportedNullIntolerant(e) => - e.children.map(IsNotNull(_): Expression).reduceLeft(And) + e.children.map(IsNotNull(_): Expression).reduceLeft(And.apply) } } } diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala index 13e3cb76805d8..c0b88798b9a35 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala @@ -64,9 +64,9 @@ object ReorderJoin extends Rule[LogicalPlan] with PredicateHelper { case (_, _) => Cross } val join = Join(left, right, innerJoinType, - joinConditions.reduceLeftOption(And), JoinHint.NONE) + joinConditions.reduceLeftOption(And.apply), JoinHint.NONE) if (others.nonEmpty) { - Filter(others.reduceLeft(And), join) + Filter(others.reduceLeft(And.apply), join) } else { join } @@ -88,7 +88,7 @@ object ReorderJoin extends Rule[LogicalPlan] with PredicateHelper { val (joinConditions, others) = conditions.partition( e => e.references.subsetOf(joinedRefs) && canEvaluateWithinJoin(e)) val joined = Join(left, right, innerJoinType, - joinConditions.reduceLeftOption(And), JoinHint.NONE) + joinConditions.reduceLeftOption(And.apply), JoinHint.NONE) // should not have reference to same logical plan createOrderedJoin(Seq((joined, Inner)) ++ rest.filterNot(_._1 eq right), others) @@ -270,11 +270,11 @@ object ExtractPythonUDFFromJoinCondition extends Rule[LogicalPlan] with Predicat log" it will be moved out and the join plan will be turned to cross join.") None } else { - Some(rest.reduceLeft(And)) + Some(rest.reduceLeft(And.apply)) } val newJoin = j.copy(condition = newCondition) joinType match { - case _: InnerLike => Filter(udf.reduceLeft(And), newJoin) + case _: InnerLike => Filter(udf.reduceLeft(And.apply), newJoin) case _ => throw QueryCompilationErrors.usePythonUDFInJoinConditionUnsupportedError(joinType) } diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala index bf3e63571dbbc..ac7cf4e64fa5d 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala @@ -152,7 +152,7 @@ object RewritePredicateSubquery extends Rule[LogicalPlan] with PredicateHelper { // Construct the pruned filter condition. val newFilter: LogicalPlan = withoutSubquery match { case Nil => child - case conditions => Filter(conditions.reduce(And), child) + case conditions => Filter(conditions.reduce(And.apply), child) } // Filter the plan by applying left semi and left anti joins. @@ -199,7 +199,7 @@ object RewritePredicateSubquery extends Rule[LogicalPlan] with PredicateHelper { // SELECT ... FROM A WHERE A.A1 NOT IN (SELECT B.B1 FROM B WHERE B.B2 = A.A2 AND B.B3 > 1) // will have the final conditions in the LEFT ANTI as // (A.A1 = B.B1 OR ISNULL(A.A1 = B.B1)) AND (B.B2 = A.A2) AND B.B3 > 1 - val finalJoinCond = (nullAwareJoinConds ++ conditions).reduceLeft(And) + val finalJoinCond = (nullAwareJoinConds ++ conditions).reduceLeft(And.apply) Join(outerPlan, rewriteDomainJoinsIfPresent(outerPlan, newSub, Some(finalJoinCond)), LeftAnti, Option(finalJoinCond), JoinHint(None, subHint)) case (p, predicate) => @@ -408,7 +408,7 @@ object RewritePredicateSubquery extends Rule[LogicalPlan] with PredicateHelper { case Exists(sub, _, _, conditions, subHint) => val exists = AttributeReference("exists", BooleanType, nullable = false)() val existenceJoin = ExistenceJoin(exists) - val newCondition = conditions.reduceLeftOption(And) + val newCondition = conditions.reduceLeftOption(And.apply) newPlan = buildJoin(newPlan, rewriteDomainJoinsIfPresent(newPlan, sub, newCondition), existenceJoin, newCondition, subHint) @@ -432,7 +432,7 @@ object RewritePredicateSubquery extends Rule[LogicalPlan] with PredicateHelper { // :- Relation[id#78,v#79] parquet // +- Relation[id#80] parquet val nullAwareJoinConds = inConditions.map(c => Or(c, IsNull(c))) - val finalJoinCond = (nullAwareJoinConds ++ conditions).reduceLeft(And) + val finalJoinCond = (nullAwareJoinConds ++ conditions).reduceLeft(And.apply) val joinHint = JoinHint(None, subHint) newPlan = Join(newPlan, rewriteDomainJoinsIfPresent(newPlan, newSub, Some(finalJoinCond)), @@ -444,7 +444,7 @@ object RewritePredicateSubquery extends Rule[LogicalPlan] with PredicateHelper { // Deduplicate conflicting attributes if any. val newSub = dedupSubqueryOnSelfJoin(newPlan, sub, Some(values)) val inConditions = values.zip(newSub.output).map(EqualTo.tupled) - val newConditions = (inConditions ++ conditions).reduceLeftOption(And) + val newConditions = (inConditions ++ conditions).reduceLeftOption(And.apply) val joinHint = JoinHint(None, subHint) newPlan = Join(newPlan, rewriteDomainJoinsIfPresent(newPlan, newSub, newConditions), ExistenceJoin(exists), newConditions, joinHint) @@ -452,7 +452,7 @@ object RewritePredicateSubquery extends Rule[LogicalPlan] with PredicateHelper { exists } } - (newExprs.reduceOption(And), newPlan, introducedAttrs.toSeq) + (newExprs.reduceOption(And.apply), newPlan, introducedAttrs.toSeq) } } @@ -493,7 +493,7 @@ object PullupCorrelatedPredicates extends Rule[LogicalPlan] with PredicateHelper correlated match { case Nil => f case xs if local.nonEmpty => - val newFilter = Filter(local.reduce(And), child) + val newFilter = Filter(local.reduce(And.apply), child) predicateMap += newFilter -> xs newFilter case xs => @@ -918,7 +918,7 @@ object RewriteCorrelatedScalarSubquery extends Rule[LogicalPlan] with AliasHelpe } lazy val planWithoutCountBug = Project( currentChild.output :+ origOutput, - Join(currentChild, query, joinType, conditions.reduceOption(And), joinHint)) + Join(currentChild, query, joinType, conditions.reduceOption(And.apply), joinHint)) if (Utils.isTesting) { assert(mayHaveCountBug.isDefined) @@ -967,7 +967,7 @@ object RewriteCorrelatedScalarSubquery extends Rule[LogicalPlan] with AliasHelpe currentChild.output :+ subqueryResultExpr, Join(currentChild, Project(query.output :+ alwaysTrueExpr, query), - joinType, conditions.reduceOption(And), joinHint)) + joinType, conditions.reduceOption(And.apply), joinHint)) } else { // CASE 3: Subquery with HAVING clause. Pull the HAVING clause above the join. @@ -999,7 +999,7 @@ object RewriteCorrelatedScalarSubquery extends Rule[LogicalPlan] with AliasHelpe currentChild.output :+ caseExpr, Join(currentChild, Project(subqueryRoot.output :+ alwaysTrueExpr, subqueryRoot), - joinType, conditions.reduceOption(And), joinHint)) + joinType, conditions.reduceOption(And.apply), joinHint)) } } } @@ -1092,7 +1092,7 @@ object RewriteLateralSubquery extends Rule[LogicalPlan] { _.containsPattern(LATERAL_JOIN)) { case LateralJoin(left, LateralSubquery(sub, _, _, joinCond, subHint), joinType, condition) => val newRight = DecorrelateInnerQuery.rewriteDomainJoins(left, sub, joinCond) - val newCond = (condition ++ joinCond).reduceOption(And) + val newCond = (condition ++ joinCond).reduceOption(And.apply) // The subquery appears on the right side of the join, hence add the hint to the right side Join(left, newRight, joinType, newCond, JoinHint(None, subHint)) } diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala index 8389e79331a96..e2d74d3c2a68c 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala @@ -3551,7 +3551,7 @@ class AstBuilder extends DataTypeAstBuilder } } else { ctx.expression.asScala.map(expression) - .map(p => invertIfNotDefined(getLike(e, p))).toSeq.reduceLeft(Or) + .map(p => invertIfNotDefined(getLike(e, p))).toSeq.reduceLeft(Or.apply) } case Some(SqlBaseParser.ALL) => if (ctx.expression.isEmpty) { @@ -3569,7 +3569,7 @@ class AstBuilder extends DataTypeAstBuilder } } else { ctx.expression.asScala.map(expression) - .map(p => invertIfNotDefined(getLike(e, p))).toSeq.reduceLeft(And) + .map(p => invertIfNotDefined(getLike(e, p))).toSeq.reduceLeft(And.apply) } case _ => val escapeChar = Option(ctx.escapeChar) @@ -7259,7 +7259,7 @@ class AstBuilder extends DataTypeAstBuilder * }}} */ override def visitCall(ctx: CallContext): LogicalPlan = withOrigin(ctx) { - val procedure = withIdentClause(ctx.identifierReference, UnresolvedProcedure) + val procedure = withIdentClause(ctx.identifierReference, UnresolvedProcedure.apply) val args = ctx.functionArgument.asScala.map { case expr if expr.namedArgumentExpression != null => val namedExpr = expr.namedArgumentExpression diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/planning/patterns.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/planning/patterns.scala index c4af18fc8739a..a8cd9a4327b2f 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/planning/patterns.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/planning/patterns.scala @@ -219,8 +219,8 @@ object ExtractEquiJoinKeys extends Logging with PredicateHelper { if (joinKeys.nonEmpty) { val (leftKeys, rightKeys) = joinKeys.unzip logDebug(s"leftKeys:$leftKeys | rightKeys:$rightKeys") - Some((joinType, leftKeys, rightKeys, otherPredicates.reduceOption(And), - predicatesOfJoinKeys.reduceOption(And), left, right, hint)) + Some((joinType, leftKeys, rightKeys, otherPredicates.reduceOption(And.apply), + predicatesOfJoinKeys.reduceOption(And.apply), left, right, hint)) } else { None } diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/NormalizePlan.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/NormalizePlan.scala index ff471cd6f00f8..a0e274b100f52 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/NormalizePlan.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/NormalizePlan.scala @@ -191,7 +191,7 @@ object NormalizePlan extends PredicateHelper { splitConjunctivePredicates(condition) .map(rewriteBinaryComparison) .sortBy(_.hashCode()) - .reduce(And), + .reduce(And.apply), child ) case sample: Sample => @@ -209,7 +209,7 @@ object NormalizePlan extends PredicateHelper { splitConjunctivePredicates(condition.get) .map(rewriteBinaryComparison) .sortBy(_.hashCode()) - .reduce(And) + .reduce(And.apply) Join(left, right, newJoinType, Some(newCondition), hint) case project: Project if normalizeProjectList => normalizeProjectListOrder(project) @@ -249,8 +249,8 @@ object NormalizePlan extends PredicateHelper { * 3. (a > b), (b < a) */ private def rewriteBinaryComparison(condition: Expression): Expression = condition match { - case EqualTo(l, r) => Seq(l, r).sortBy(_.hashCode()).reduce(EqualTo) - case EqualNullSafe(l, r) => Seq(l, r).sortBy(_.hashCode()).reduce(EqualNullSafe) + case EqualTo(l, r) => Seq(l, r).sortBy(_.hashCode()).reduce(EqualTo.apply) + case EqualNullSafe(l, r) => Seq(l, r).sortBy(_.hashCode()).reduce(EqualNullSafe.apply) case GreaterThan(l, r) if l.hashCode() > r.hashCode() => LessThan(r, l) case LessThan(l, r) if l.hashCode() > r.hashCode() => GreaterThan(r, l) case GreaterThanOrEqual(l, r) if l.hashCode() > r.hashCode() => LessThanOrEqual(r, l) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala index 8ca3f76c3b885..4983b0997cac0 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala @@ -826,7 +826,7 @@ object QueryPlan extends PredicateHelper { */ def normalizePredicates(predicates: Seq[Expression], output: AttributeSeq): Seq[Expression] = { if (predicates.nonEmpty) { - val normalized = normalizeExpressions(predicates.reduce(And), output) + val normalized = normalizeExpressions(predicates.reduce(And.apply), output) splitConjunctivePredicates(normalized) } else { Nil diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/QueryPlanConstraints.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/QueryPlanConstraints.scala index b023d7320808d..746ae2d6fa9c8 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/QueryPlanConstraints.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/QueryPlanConstraints.scala @@ -162,7 +162,7 @@ trait ConstraintHelper { // Second, we infer additional constraints from non-nullable attributes that are part of the // operator's output val nonNullableAttributes = output.filterNot(_.nullable) - isNotNullConstraints ++= nonNullableAttributes.map(IsNotNull) + isNotNullConstraints ++= nonNullableAttributes.map(IsNotNull.apply) isNotNullConstraints -- constraints } diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala index b7c441063874d..7879a85bc899a 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala @@ -537,7 +537,7 @@ abstract class UnionBase extends LogicalPlan { val otherb = b.diff(common).filter(_.references.size == 1).groupBy(_.references.head) // loose the constraints by: A1 && B1 || A2 && B2 -> (A1 || A2) && (B1 || B2) val others = (othera.keySet intersect otherb.keySet).map { attr => - Or(othera(attr).reduceLeft(And), otherb(attr).reduceLeft(And)) + Or(othera(attr).reduceLeft(And.apply), otherb(attr).reduceLeft(And.apply)) } common ++ others } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/columnar/CachedBatchSerializer.scala b/sql/core/src/main/scala/org/apache/spark/sql/columnar/CachedBatchSerializer.scala index 2f1b79ec90f99..1935a647a5829 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/columnar/CachedBatchSerializer.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/columnar/CachedBatchSerializer.scala @@ -354,7 +354,7 @@ abstract class SimpleMetricsCachedBatchSerializer extends CachedBatchSerializer def ret(index: Int, cachedBatchIterator: Iterator[CachedBatch]): Iterator[CachedBatch] = { val partitionFilter = Predicate.create( - partitionFilters.reduceOption(And).getOrElse(Literal(true)), + partitionFilters.reduceOption(And.apply).getOrElse(Literal(true)), cachedAttributes) partitionFilter.initialize(index) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala index be7013188f2f9..47390702d09fd 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala @@ -395,7 +395,7 @@ trait FileSourceScanLike extends DataSourceScanExec with SessionStateHelper { if (dynamicPartitionFilters.nonEmpty) { val startTime = System.nanoTime() // call the file index for the files matching all filters except dynamic partition filters - val predicate = dynamicPartitionFilters.reduce(And) + val predicate = dynamicPartitionFilters.reduce(And.apply) val partitionColumns = relation.partitionSchema val boundPredicate = Predicate.create(predicate.transform { case a: AttributeReference => diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlanner.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlanner.scala index 96a3e3179f541..3f0b3ca69bef1 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlanner.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlanner.scala @@ -97,7 +97,7 @@ class SparkPlanner(val session: SparkSession, val experimentalMethods: Experimen val projectSet = AttributeSet(projectList.flatMap(_.references)) val filterSet = AttributeSet(filterPredicates.flatMap(_.references)) val filterCondition: Option[Expression] = - prunePushedDownFilters(filterPredicates).reduceLeftOption(And) + prunePushedDownFilters(filterPredicates).reduceLeftOption(And.apply) // Right now we still use a projection even if the only evaluation is applying an alias // to a column. Since this is a no-op, it could be avoided. However, using this diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala index 6e761fbe07b27..f0767d6414645 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala @@ -479,7 +479,7 @@ abstract class SparkStrategies extends QueryPlanner[SparkPlan] { case other => residuals += other } - val residual = residuals.reduceOption(And) + val residual = residuals.reduceOption(And.apply) (leftKeys.toSeq, rightKeys.toSeq, residual) } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/DescribeRelationJsonCommand.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/DescribeRelationJsonCommand.scala index 8746fdbc92499..f5489232c1f70 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/DescribeRelationJsonCommand.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/DescribeRelationJsonCommand.scala @@ -150,7 +150,7 @@ case class DescribeRelationJsonCommand( addKeyValueToMap("table_name", JString(ident.last), jsonMap) addKeyValueToMap("catalog_name", JString(ident.head), jsonMap) val namespace = ident.init.tail - addKeyValueToMap("namespace", JArray(namespace.map(JString).toList), jsonMap) + addKeyValueToMap("namespace", JArray(namespace.map(JString.apply).toList), jsonMap) if (namespace.nonEmpty) { addKeyValueToMap("schema_name", JString(namespace.last), jsonMap) } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala index 05bc14dfb700f..8c8b15b06a141 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala @@ -486,7 +486,7 @@ object DataSourceStrategy // Combines all Catalyst filter `Expression`s that are either not convertible to data source // `Filter`s or cannot be handled by `relation`. - val filterCondition = unhandledPredicates.reduceLeftOption(expressions.And) + val filterCondition = unhandledPredicates.reduceLeftOption(expressions.And.apply) if (projects.map(_.toAttribute) == projects && projectSet.size == projects.size && @@ -754,7 +754,7 @@ object DataSourceStrategy case expressions.Not(child) => translateFilterWithMapping(child, translatedFilterToExpr, nestedPredicatePushdownEnabled) - .map(sources.Not) + .map(sources.Not.apply) case other => val filter = translateLeafNodeFilter(other, PushableColumn(nestedPredicatePushdownEnabled)) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileIndex.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileIndex.scala index 0291a5fd28a72..56b4d099a7c0b 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileIndex.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileIndex.scala @@ -61,7 +61,7 @@ class FilePruningRunner(filters: Seq[Expression]) { metadataAttr.name != FileFormat.FILE_BLOCK_LENGTH case _ => false } - }.reduceOption(And) + }.reduceOption(And.apply) // - Retrieve all required metadata attributes and put them into a sequence // - Bind all file constant metadata attribute references to their respective index diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala index b7a1736bc2e9e..156a542e5c408 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala @@ -134,7 +134,7 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging { val numBuckets = bucketSpec.numBuckets val normalizedFiltersAndExpr = normalizedFilters - .reduce(expressions.And) + .reduce(expressions.And.apply) val matchedBuckets = getExpressionBuckets(normalizedFiltersAndExpr, bucketColumnName, numBuckets) @@ -357,7 +357,8 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging { }.getOrElse(scan) // bottom-most filters are put in the left of the list. - val finalFilters = afterScanFilters.toSeq.reduceOption(expressions.And).toSeq ++ stayUpFilters + val finalFilters = + afterScanFilters.toSeq.reduceOption(expressions.And.apply).toSeq ++ stayUpFilters val withFilter = finalFilters.foldLeft(withMetadataProjections)((plan, cond) => { execution.FilterExec(cond, plan) }) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningAwareFileIndex.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningAwareFileIndex.scala index 7e6b733e6aafc..7c3ae10294c45 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningAwareFileIndex.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningAwareFileIndex.scala @@ -194,7 +194,7 @@ abstract class PartitioningAwareFileIndex( } if (partitionPruningPredicates.nonEmpty) { - val predicate = partitionPruningPredicates.reduce(expressions.And) + val predicate = partitionPruningPredicates.reduce(expressions.And.apply) val boundPredicate = Predicate.createInterpreted(predicate.transform { case a: AttributeReference => diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PruneFileSourcePartitions.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PruneFileSourcePartitions.scala index 6af6dc721a6f8..d2228f3a54c9f 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PruneFileSourcePartitions.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PruneFileSourcePartitions.scala @@ -39,7 +39,7 @@ private[sql] object PruneFileSourcePartitions extends Rule[LogicalPlan] { filters: Seq[Expression], relation: LeafNode): Project = { val withFilter = if (filters.nonEmpty) { - val filterExpression = filters.reduceLeft(And) + val filterExpression = filters.reduceLeft(And.apply) Filter(filterExpression, relation) } else { relation @@ -78,7 +78,7 @@ private[sql] object PruneFileSourcePartitions extends Rule[LogicalPlan] { fsRelation.copy(location = prunedFileIndex)(fsRelation.sparkSession) // Change table stats based on the sizeInBytes of pruned files val filteredStats = - FilterEstimation(Filter(partitionKeyFilters.reduce(And), logicalRelation)).estimate + FilterEstimation(Filter(partitionKeyFilters.reduce(And.apply), logicalRelation)).estimate val colStats = filteredStats.map(_.attributeStats.map { case (attr, colStat) => (attr.name, colStat.toCatalogColumnStat(attr.name, attr.dataType)) }) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PushVariantIntoScan.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PushVariantIntoScan.scala index 69aa576f0c94d..811cc4d32562f 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PushVariantIntoScan.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PushVariantIntoScan.scala @@ -421,7 +421,7 @@ object PushVariantIntoScan extends Rule[LogicalPlan] { val newRelation = relation.copy(relation = newHadoopFsRelation, output = newOutput.toIndexedSeq) val withFilter = if (filters.nonEmpty) { - Filter(filters.map(variants.rewriteExpr(_, attributeMap)).reduce(And), newRelation) + Filter(filters.map(variants.rewriteExpr(_, attributeMap)).reduce(And.apply), newRelation) } else { newRelation } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SchemaPruning.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SchemaPruning.scala index 202599a819d8d..074d7e00f8d5d 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SchemaPruning.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SchemaPruning.scala @@ -42,7 +42,7 @@ object SchemaPruning extends Rule[LogicalPlan] { plan transformDown { case op @ ScanOperation(projects, filtersStayUp, filtersPushDown, l @ LogicalRelationWithTable(hadoopFsRelation: HadoopFsRelation, _)) => - val allFilters = filtersPushDown.reduceOption(And).toSeq ++ filtersStayUp + val allFilters = filtersPushDown.reduceOption(And.apply).toSeq ++ filtersStayUp prunePhysicalColumns(l, projects, allFilters, hadoopFsRelation, (prunedDataSchema, prunedMetadataSchema) => { val prunedHadoopRelation = diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilters.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilters.scala index 66770870dabe9..da6e4ad15bfbd 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilters.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilters.scala @@ -127,7 +127,7 @@ private[sql] object OrcFilters extends OrcFiltersBase { } yield Or(lhs, rhs) case Not(pred) => val childResultOptional = convertibleFiltersHelper(pred, canPartialPushDown = false) - childResultOptional.map(Not) + childResultOptional.map(Not.apply) case other => for (_ <- buildLeafSearchArgument(dataTypeMap, other, newBuilder())) yield other } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala index f60ced3eb5973..4c12ed864170f 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala @@ -626,7 +626,7 @@ class ParquetFilters( } case sources.Not(pred) => val resultOptional = convertibleFiltersHelper(pred, canPartialPushDown = false) - resultOptional.map(sources.Not) + resultOptional.map(sources.Not.apply) case other => if (createFilter(other).isDefined) { diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala index 9c745461b1709..aac4fb980c876 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala @@ -994,7 +994,7 @@ private[sql] object DataSourceV2Strategy extends Logging { filters: Seq[Expression], scan: LeafExecNode, needsUnsafeConversion: Boolean): SparkPlan = { - val filterCondition = filters.reduceLeftOption(And) + val filterCondition = filters.reduceLeftOption(And.apply) val withFilter = filterCondition.map(FilterExec(_, scan)).getOrElse(scan) if (withFilter.output != project || needsUnsafeConversion) { diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupBasedRowLevelOperationScanPlanning.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupBasedRowLevelOperationScanPlanning.scala index 0cb7967cb7369..4033d93d9360b 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupBasedRowLevelOperationScanPlanning.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupBasedRowLevelOperationScanPlanning.scala @@ -133,6 +133,6 @@ object GroupBasedRowLevelOperationScanPlanning extends Rule[LogicalPlan] with Pr val evaluatedFilterSet = ExpressionSet(evaluatedFilters) val predicates = splitConjunctivePredicates(cond) val remainingPredicates = predicates.filterNot(evaluatedFilterSet.contains) - remainingPredicates.reduceLeftOption(And).getOrElse(TrueLiteral) + remainingPredicates.reduceLeftOption(And.apply).getOrElse(TrueLiteral) } } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala index 14d9f754f95c5..42a3eb73f3653 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala @@ -125,7 +125,7 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper { |Post-Scan Filters: ${MDC(POST_SCAN_FILTERS, postScanFilters.mkString(","))} """.stripMargin) - val filterCondition = postScanFilters.reduceLeftOption(And) + val filterCondition = postScanFilters.reduceLeftOption(And.apply) filterCondition.map(Filter(_, sHolder)).getOrElse(sHolder) } @@ -895,7 +895,7 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper { // Rewrite filter expressions using the variant transformation val rewrittenFilters = if (filters.nonEmpty) { val rewrittenFilterExprs = filters.map(variants.rewriteExpr(_, attributeMap)) - Some(rewrittenFilterExprs.reduce(And)) + Some(rewrittenFilterExprs.reduce(And.apply)) } else { None } @@ -923,7 +923,7 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper { val normalizedProjects = DataSourceStrategy .normalizeExprs(project, sHolder.output) .asInstanceOf[Seq[NamedExpression]] - val allFilters = filtersPushDown.reduceOption(And).toSeq ++ filtersStayUp + val allFilters = filtersPushDown.reduceOption(And.apply).toSeq ++ filtersStayUp val normalizedFilters = DataSourceStrategy.normalizeExprs(allFilters, sHolder.output) val (scan, output) = PushDownUtils.pruneColumns( sHolder.builder, sHolder.relation, normalizedProjects, normalizedFilters) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/join/StreamingSymmetricHashJoinHelper.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/join/StreamingSymmetricHashJoinHelper.scala index e96c9dbfd7e1b..e43ddb072aa1a 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/join/StreamingSymmetricHashJoinHelper.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/join/StreamingSymmetricHashJoinHelper.scala @@ -143,10 +143,10 @@ object StreamingSymmetricHashJoinHelper extends Logging { } ( - leftConjuncts.reduceOption(And), - rightConjuncts.reduceOption(And), + leftConjuncts.reduceOption(And.apply), + rightConjuncts.reduceOption(And.apply), (nonLeftConjuncts.intersect(nonRightConjuncts) ++ nonDeterministicConjuncts) - .reduceOption(And) + .reduceOption(And.apply) ) } } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/sources/TextSocketMicroBatchStream.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/sources/TextSocketMicroBatchStream.scala index 243d6c784f5ae..bd12dce1e08c5 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/sources/TextSocketMicroBatchStream.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/sources/TextSocketMicroBatchStream.scala @@ -128,7 +128,7 @@ class TextSocketMicroBatchStream(host: String, port: Int, numPartitions: Int) slices(idx % numPartitions).append(r) } - slices.map(TextSocketInputPartition) + slices.map(TextSocketInputPartition.apply) } override def createReaderFactory(): PartitionReaderFactory = diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala index 8284f50ec1a8f..07edeece23777 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala @@ -800,7 +800,7 @@ class RocksDBFileManager( // Delete orphan checksum files val checksumFiles = allFiles .filter(ChecksumCheckpointFileManager.isChecksumFile) - .map(ChecksumFile) + .map(ChecksumFile.apply) // Do this only if we see checksum files in the initial dir listing // To avoid checking for orphan checksum files, if there is no checksum files diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveTableScanExec.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveTableScanExec.scala index 5e4ef82d798b8..b2ecd3bb94163 100644 --- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveTableScanExec.scala +++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveTableScanExec.scala @@ -81,7 +81,7 @@ case class HiveTableScanExec( // Bind all partition key attribute references in the partition pruning predicate for later // evaluation. - private lazy val boundPruningPred = partitionPruningPred.reduceLeftOption(And).map { pred => + private lazy val boundPruningPred = partitionPruningPred.reduceLeftOption(And.apply).map { pred => require(pred.dataType == BooleanType, s"Data type of predicate $pred must be ${BooleanType.catalogString} rather than " + s"${pred.dataType.catalogString}.") diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/PruneHiveTablePartitions.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/PruneHiveTablePartitions.scala index 6486904fe65af..7eeec5032d0fa 100644 --- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/PruneHiveTablePartitions.scala +++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/PruneHiveTablePartitions.scala @@ -84,7 +84,7 @@ private[sql] class PruneHiveTablePartitions(session: SparkSession) } if (sizeOfPartitions.forall(_ > 0)) { val filteredStats = - FilterEstimation(Filter(partitionKeyFilters.reduce(And), relation)).estimate + FilterEstimation(Filter(partitionKeyFilters.reduce(And.apply), relation)).estimate val colStats = filteredStats.map(_.attributeStats.map { case (attr, colStat) => (attr.name, colStat.toCatalogColumnStat(attr.name, attr.dataType)) }) @@ -114,7 +114,7 @@ private[sql] class PruneHiveTablePartitions(session: SparkSession) val newRelation = relation.copy( tableMeta = newTableMeta, prunedPartitions = Some(newPartitions)) // Keep partition filters so that they are visible in physical planning - Project(projections, Filter(filters.reduceLeft(And), newRelation)) + Project(projections, Filter(filters.reduceLeft(And.apply), newRelation)) } else { op }