svn commit: r1911036 - in /pig/trunk: CHANGES.txt src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/PigGraceShuffleVertexManager.java

[rohini]

Author: rohini
Date: Sat Jul 15 20:03:23 2023
New Revision: 1911036

URL: http://svn.apache.org/viewvc?rev=1911036&view=rev
Log:
PIG-5441: Pig skew join tez grace reducer fails to find shuffle data (yigress 
via rohini)

Modified:
    pig/trunk/CHANGES.txt
    
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/PigGraceShuffleVertexManager.java

Modified: pig/trunk/CHANGES.txt
URL: 
http://svn.apache.org/viewvc/pig/trunk/CHANGES.txt?rev=1911036&r1=1911035&r2=1911036&view=diff
==============================================================================
--- pig/trunk/CHANGES.txt (original)
+++ pig/trunk/CHANGES.txt Sat Jul 15 20:03:23 2023
@@ -133,6 +133,8 @@ OPTIMIZATIONS
  
 BUG FIXES
 
+PIG-5441: Pig skew join tez grace reducer fails to find shuffle data (yigress 
via rohini)
+
 PIG-5432: OrcStorage fails to detect schema in some cases (jtolar via rohini)
 
 PIG-5439: pig.exec.reducers.max does not take effect for skewed join 
(vnarayanan7 via rohini)

Modified: 
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/PigGraceShuffleVertexManager.java
URL: 
http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/PigGraceShuffleVertexManager.java?rev=1911036&r1=1911035&r2=1911036&view=diff
==============================================================================
--- 
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/PigGraceShuffleVertexManager.java
 (original)
+++ 
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/PigGraceShuffleVertexManager.java
 Sat Jul 15 20:03:23 2023
@@ -165,7 +165,8 @@ public class PigGraceShuffleVertexManage
             Map<String, EdgeProperty> edgeManagers = new HashMap<String, 
EdgeProperty>();
             for(Map.Entry<String,EdgeProperty> entry : 
getContext().getInputVertexEdgeProperties().entrySet()) {
                 EdgeProperty edge = entry.getValue();
-                edge = EdgeProperty.create(DataMovementType.SCATTER_GATHER, 
edge.getDataSourceType(), edge.getSchedulingType(),
+                DataMovementType movementType = edge.getDataMovementType() == 
DataMovementType.BROADCAST? DataMovementType.BROADCAST: 
DataMovementType.SCATTER_GATHER;
+                edge = EdgeProperty.create(movementType, 
edge.getDataSourceType(), edge.getSchedulingType(),
                         edge.getEdgeSource(), edge.getEdgeDestination());
                 edgeManagers.put(entry.getKey(), edge);
             }