Update to 0.37.9-public

Author: cornmonster

odps-sdk-impl/odps-mapred-bridge/src/main/java/com/aliyun/odps/mapred/bridge/type/ColumnBasedRecordComparator.java

@@ -23,7 +23,6 @@
 import java.util.Map;
 
 import com.aliyun.odps.Column;
-import com.aliyun.odps.mapred.conf.JobConf;
 import com.aliyun.odps.mapred.conf.JobConf.SortOrder;
 
 /**
@@ -32,7 +31,7 @@
 public class ColumnBasedRecordComparator extends NaturalRecordComparator {
 
   int[] selIdxs;
-  SortOrder[] sortOrder;
+  SortOrder[] sortOrders;
 
   /**
    * ColumnBasedRecordComparator的构造方法，指定排序列的下标索引。例如指定排序列的下标索引为0，3，则比较时先用第0列比较，
@@ -46,7 +45,7 @@ public class ColumnBasedRecordComparator extends NaturalRecordComparator {
   public ColumnBasedRecordComparator(int[] selIdxs, Column[] schema) {
     super(schema);
     this.selIdxs = selIdxs;
-    this.sortOrder = null;
+    this.sortOrders = null;
   }
 
   /**
@@ -60,11 +59,17 @@ public ColumnBasedRecordComparator(String[] selCols, Column[] schema) {
     this(selCols, schema, null);
   }
 
-  public ColumnBasedRecordComparator(String[] selCols, Column[] schema, SortOrder[] sortOrder) {
+  public ColumnBasedRecordComparator(String[] selCols, Column[] schema, SortOrder[] sortOrders) {
     super(schema);
-    this.sortOrder = sortOrder;
+
+    if (sortOrders != null && selCols.length != sortOrders.length) {
+      throw new IllegalArgumentException(
+          "Number of sort column is not equal to length of sort order array.");
+    }
+
+    this.sortOrders = sortOrders;
     this.selIdxs = new int[selCols.length];
-    Map<String, Integer> reverseLookupMap = new HashMap<String, Integer>();
+    Map<String, Integer> reverseLookupMap = new HashMap<>();
     int i = 0;
     for (Column c : schema) {
       reverseLookupMap.put(c.getName(), i);
@@ -80,21 +85,29 @@ public ColumnBasedRecordComparator(String[] selCols, Column[] schema, SortOrder[
   @Override
   public int compare(Object[] l, Object[] r) {
     int result = 0;
+    int index = 0;
     for (int i : selIdxs) {
       if (r.length < i) {
+        if (sortOrders != null && sortOrders[index] == SortOrder.DESC) {
+          return 1;
+        }
         return -1;
       }
       if (l.length < i) {
+        if (sortOrders != null && sortOrders[index] == SortOrder.DESC) {
+          return -1;
+        }
         return 1;
       }
-      if (sortOrder != null && sortOrder[i] == JobConf.SortOrder.DESC) {
-        result = compare(r[i], l[i], comparators[i]);
-      } else {
-        result = compare(l[i], r[i], comparators[i]);
-      }
+      result = compare(l[i], r[i], comparators[i]);
       if (result != 0) {
+        if (sortOrders != null && sortOrders.length == selIdxs.length
+            && sortOrders[index] == SortOrder.DESC) {
+          return (0 - result);
+        }
         return result;
       }
+      index++;
     }
     return 0;
   }