[spark] branch master updated: [SPARK-44361][SQL][FOLLOWUP] Use PartitionEvaluator API in MapInBatchExec

[wenchen]

This is an automated email from the ASF dual-hosted git repository.

wenchen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/master by this push:
     new 3cf88cb6c42 [SPARK-44361][SQL][FOLLOWUP] Use PartitionEvaluator API in 
MapInBatchExec
3cf88cb6c42 is described below

commit 3cf88cb6c42e802fc4828c397df61623663be9b0
Author: Vinod KC <vinod.kc...@gmail.com>
AuthorDate: Fri Jul 28 12:49:13 2023 +0800

    [SPARK-44361][SQL][FOLLOWUP] Use PartitionEvaluator API in MapInBatchExec
    
    ### What changes were proposed in this pull request?
    This is a follow-up of https://github.com/apache/spark/pull/42024, to set 
the partition index correctly even if it's not used for now.
    
    ### Why are the changes needed?
    future-proof
    
    ### Does this PR introduce _any_ user-facing change?
    
    No
    
    ### How was this patch tested?
    
    existing tests
    
    Closes #42189 from vinodkc/br_SPARK-44361_Followup.
    
    Authored-by: Vinod KC <vinod.kc...@gmail.com>
    Signed-off-by: Wenchen Fan <wenc...@databricks.com>
---
 .../org/apache/spark/sql/execution/python/MapInBatchExec.scala    | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git 
a/sql/core/src/main/scala/org/apache/spark/sql/execution/python/MapInBatchExec.scala
 
b/sql/core/src/main/scala/org/apache/spark/sql/execution/python/MapInBatchExec.scala
index 4a47c2089d6..8db389f0266 100644
--- 
a/sql/core/src/main/scala/org/apache/spark/sql/execution/python/MapInBatchExec.scala
+++ 
b/sql/core/src/main/scala/org/apache/spark/sql/execution/python/MapInBatchExec.scala
@@ -65,8 +65,8 @@ trait MapInBatchExec extends UnaryExecNode with 
PythonSQLMetrics {
       if (conf.usePartitionEvaluator) {
         rddBarrier.mapPartitionsWithEvaluator(evaluatorFactory)
       } else {
-        rddBarrier.mapPartitions { iter =>
-          evaluatorFactory.createEvaluator().eval(0, iter)
+        rddBarrier.mapPartitionsWithIndex { (index, iter) =>
+          evaluatorFactory.createEvaluator().eval(index, iter)
         }
       }
     } else {
@@ -74,8 +74,8 @@ trait MapInBatchExec extends UnaryExecNode with 
PythonSQLMetrics {
       if (conf.usePartitionEvaluator) {
         inputRdd.mapPartitionsWithEvaluator(evaluatorFactory)
       } else {
-        inputRdd.mapPartitionsInternal { iter =>
-          evaluatorFactory.createEvaluator().eval(0, iter)
+        inputRdd.mapPartitionsWithIndexInternal { (index, iter) =>
+          evaluatorFactory.createEvaluator().eval(index, iter)
         }
       }
     }


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org