From 0252663f998edd1deb75e99b9a1d7ab9ec806a29 Mon Sep 17 00:00:00 2001 From: Liang-Chi Hsieh Date: Tue, 26 Sep 2017 02:11:02 +0000 Subject: [PATCH] Claim deferred input evaluation for Sample and Limit. --- .../apache/spark/sql/execution/basicPhysicalOperators.scala | 4 ++++ .../src/main/scala/org/apache/spark/sql/execution/limit.scala | 4 ++++ 2 files changed, 8 insertions(+) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala index 18142c44f029..8389e2f3d5be 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala @@ -267,6 +267,10 @@ case class SampleExec( } } + // Mark this as empty. This plan doesn't need to evaluate any inputs and can defer the evaluation + // to the parent operator. + override def usedInputs: AttributeSet = AttributeSet.empty + override def inputRDDs(): Seq[RDD[InternalRow]] = { child.asInstanceOf[CodegenSupport].inputRDDs() } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/limit.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/limit.scala index 73a0f8735ed4..1f515e29b4af 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/limit.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/limit.scala @@ -62,6 +62,10 @@ trait BaseLimitExec extends UnaryExecNode with CodegenSupport { child.asInstanceOf[CodegenSupport].inputRDDs() } + // Mark this as empty. This plan doesn't need to evaluate any inputs and can defer the evaluation + // to the parent operator. + override def usedInputs: AttributeSet = AttributeSet.empty + protected override def doProduce(ctx: CodegenContext): String = { child.asInstanceOf[CodegenSupport].produce(ctx, this) }