From 600c9bf46fc7c715addc3f8d7f1e0d334d9db14e Mon Sep 17 00:00:00 2001 From: Wenchen Fan Date: Tue, 3 Jan 2023 20:26:24 +0800 Subject: [PATCH] move expression initialization code to the base class --- .../sql/catalyst/expressions/ExpressionsEvaluator.scala | 7 +++++++ .../expressions/InterpretedMutableProjection.scala | 5 +---- .../catalyst/expressions/InterpretedSafeProjection.scala | 5 +---- .../catalyst/expressions/InterpretedUnsafeProjection.scala | 5 +---- .../apache/spark/sql/catalyst/expressions/Projection.scala | 5 +---- .../apache/spark/sql/catalyst/expressions/predicates.scala | 6 +----- 6 files changed, 12 insertions(+), 21 deletions(-) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExpressionsEvaluator.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExpressionsEvaluator.scala index dcbc6926cd335..1fc0144fede07 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExpressionsEvaluator.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExpressionsEvaluator.scala @@ -42,4 +42,11 @@ trait ExpressionsEvaluator { * The default implementation does nothing. */ def initialize(partitionIndex: Int): Unit = {} + + protected def initializeExprs(exprs: Seq[Expression], partitionIndex: Int): Unit = { + exprs.foreach(_.foreach { + case n: Nondeterministic => n.initialize(partitionIndex) + case _ => + }) + } } diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/InterpretedMutableProjection.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/InterpretedMutableProjection.scala index 682604b9bf72c..01e9de085dad6 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/InterpretedMutableProjection.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/InterpretedMutableProjection.scala @@ -41,10 +41,7 @@ class InterpretedMutableProjection(expressions: Seq[Expression]) extends Mutable private[this] val buffer = new Array[Any](expressions.size) override def initialize(partitionIndex: Int): Unit = { - exprs.foreach(_.foreach { - case n: Nondeterministic => n.initialize(partitionIndex) - case _ => - }) + initializeExprs(exprs, partitionIndex) } private[this] val validExprs = expressions.zipWithIndex.filter { diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/InterpretedSafeProjection.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/InterpretedSafeProjection.scala index 84263d97f5da7..87539e80b0bd8 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/InterpretedSafeProjection.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/InterpretedSafeProjection.scala @@ -101,10 +101,7 @@ class InterpretedSafeProjection(expressions: Seq[Expression]) extends Projection } override def initialize(partitionIndex: Int): Unit = { - expressions.foreach(_.foreach { - case n: Nondeterministic => n.initialize(partitionIndex) - case _ => - }) + initializeExprs(exprs, partitionIndex) } override def apply(row: InternalRow): InternalRow = { diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/InterpretedUnsafeProjection.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/InterpretedUnsafeProjection.scala index 9108a045c0920..90a9044469510 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/InterpretedUnsafeProjection.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/InterpretedUnsafeProjection.scala @@ -67,10 +67,7 @@ class InterpretedUnsafeProjection(expressions: Array[Expression]) extends Unsafe } override def initialize(partitionIndex: Int): Unit = { - exprs.foreach(_.foreach { - case n: Nondeterministic => n.initialize(partitionIndex) - case _ => - }) + initializeExprs(exprs, partitionIndex) } override def apply(row: InternalRow): UnsafeRow = { diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Projection.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Projection.scala index 20969fa584a87..7d993d776d159 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Projection.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Projection.scala @@ -41,10 +41,7 @@ class InterpretedProjection(expressions: Seq[Expression]) extends Projection { } override def initialize(partitionIndex: Int): Unit = { - exprArray.foreach(_.foreach { - case n: Nondeterministic => n.initialize(partitionIndex) - case _ => - }) + initializeExprs(exprArray, partitionIndex) } def apply(input: InternalRow): InternalRow = { diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/predicates.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/predicates.scala index 4e4ac6ee49265..6a58f8d341682 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/predicates.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/predicates.scala @@ -53,11 +53,7 @@ case class InterpretedPredicate(expression: Expression) extends BasePredicate { } override def initialize(partitionIndex: Int): Unit = { - super.initialize(partitionIndex) - expr.foreach { - case n: Nondeterministic => n.initialize(partitionIndex) - case _ => - } + initializeExprs(Seq(expr), partitionIndex) } }