svn commit: r1791874 - in /pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/converter: IndexedKey.java PigSecondaryKeyComparatorSpark.java

[zl...@apache.org]

Author: zly
Date: Wed Apr 19 08:22:02 2017
New Revision: 1791874

URL: http://svn.apache.org/viewvc?rev=1791874&view=rev
Log:
PIG-5171:SecondarySort_7 is failing with spark exec type(Liyun)

Modified:
    pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/converter/IndexedKey.java
    pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/converter/PigSecondaryKeyComparatorSpark.java

Modified: pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/converter/IndexedKey.java
URL: http://svn.apache.org/viewvc/pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/converter/IndexedKey.java?rev=1791874&r1=1791873&r2=1791874&view=diff
==============================================================================
--- pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/converter/IndexedKey.java (original)
+++ pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/converter/IndexedKey.java Wed Apr 19 08:22:02 2017
@@ -161,8 +161,8 @@ public class IndexedKey implements Seria
             if (useSecondaryKey) {
                 Tuple thisCompoundKey = (Tuple) key;
                 Tuple thatCompoundKey = (Tuple)that.getKey();
-                return PigSecondaryKeyComparatorSpark.compareKeys(thisCompoundKey, thatCompoundKey,
-                        secondarySortOrder);
+                PigSecondaryKeyComparatorSpark comparator = new PigSecondaryKeyComparatorSpark(secondarySortOrder);
+                return comparator.compareCompoundKey(thisCompoundKey, thatCompoundKey);
             } else {
                 return DataType.compare(key, that.getKey());
             }

Modified: pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/converter/PigSecondaryKeyComparatorSpark.java
URL: http://svn.apache.org/viewvc/pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/converter/PigSecondaryKeyComparatorSpark.java?rev=1791874&r1=1791873&r2=1791874&view=diff
==============================================================================
--- pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/converter/PigSecondaryKeyComparatorSpark.java (original)
+++ pig/branches/spark/src/org/apache/pig/backend/hadoop/executionengine/spark/converter/PigSecondaryKeyComparatorSpark.java Wed Apr 19 08:22:02 2017
@@ -39,6 +39,9 @@ class PigSecondaryKeyComparatorSpark imp
         secondarySortOrder = pSecondarySortOrder;
     }
 
+    //IndexedKeyPartitioner will put the tuple with same mainKey together, in PigSecondaryKeyComparatorSpark#compare
+    // (Object o1, Object o2)
+    //we only compare the secondaryKey
     @Override
     public int compare(Object o1, Object o2) {
         Tuple t1 = (Tuple) o1;
@@ -56,7 +59,7 @@ class PigSecondaryKeyComparatorSpark imp
             }
             Object secondaryKey1 = compoundKey1.get(1);
             Object secondaryKey2 = compoundKey2.get(1);
-            int res = compareSecondaryKeys(secondaryKey1, secondaryKey2, secondarySortOrder);
+            int res = compareKeys(secondaryKey1, secondaryKey2, secondarySortOrder);
             if (LOG.isDebugEnabled()) {
                 LOG.debug("t1:" + t1 + "t2:" + t2 + " res:" + res);
             }
@@ -66,11 +69,33 @@ class PigSecondaryKeyComparatorSpark imp
         }
     }
 
-    private int compareSecondaryKeys(Object o1, Object o2, boolean[] asc){
-        return compareKeys(o1, o2, asc);
+    //compare the mainKey and secondaryKey
+    public int compareCompoundKey(Tuple compoundKey1, Tuple compoundKey2){
+        try {
+            if ((compoundKey1.size() < 2) || (compoundKey2.size() < 2)) {
+                throw new RuntimeException("compoundKey size must bigger than, compoundKey[0] stands for firstKey," +
+                        "compoundKey[1] stands for secondaryKey");
+            }
+            Object mainKey1 = compoundKey1.get(0);
+            Object mainKey2 = compoundKey2.get(0);
+            int res = compareKeys(mainKey1,mainKey2, null);
+            if ( res !=0 ){
+                return res;
+            } else {
+                Object secondaryKey1 = compoundKey1.get(1);
+                Object secondaryKey2 = compoundKey2.get(1);
+                res = compareKeys(secondaryKey1, secondaryKey2, secondarySortOrder);
+                if (LOG.isDebugEnabled()) {
+                    LOG.debug("compoundKey1:" + compoundKey1 + "compoundKey2:" + compoundKey2 + " res:" + res);
+                }
+                return res;
+            }
+        } catch (ExecException e) {
+            throw new RuntimeException("Fail to get the compoundKey", e);
+        }
     }
 
-    public static int compareKeys(Object o1, Object o2, boolean[] asc) {
+    private int compareKeys(Object o1, Object o2, boolean[] asc) {
         int rc = 0;
         if (o1 != null && o2 != null && o1 instanceof Tuple && o2 instanceof Tuple) {
             // objects are Tuples, we may need to apply sort order inside them