From 5d2956015cbe2d0f73650fd2f1f369eaebbee114 Mon Sep 17 00:00:00 2001
From: Michael Munday <mike.munday@ibm.com>
Date: Wed, 24 Jun 2020 15:43:36 +0100
Subject: [PATCH] [SPARK-32485][SQL][TEST] Fix RecordBinaryComparator tests on
 big-endian platforms.

Comparisons performed by RecordBinaryComparator are executed in
lexicographic order (byte by byte starting from the byte at index 0).
This means that the underlying endianness of the platform can affect
the result of a comparison between multi-byte values such as longs.
This difference means that two tests fail on big-endian platforms.

Also, the two tests compare 'special' long values to test edge cases
that triggered old bugs in the fast path of the comparator. However
since PR #26548 these 'special' values are byte-reversed before being
compared on little-endian platforms, which means that the edge cases
these 'special' values were designed to trigger are no longer tested.

This PR fixes both these issues by byte reversing the values on
little-endian systems.

RecordBinaryComparatorSuite tests now pass on a big-endian machine
(s390x).
---
 .../sort/RecordBinaryComparatorSuite.java     | 48 ++++++++++++++++---
 1 file changed, 42 insertions(+), 6 deletions(-)

diff --git a/sql/core/src/test/java/test/org/apache/spark/sql/execution/sort/RecordBinaryComparatorSuite.java b/sql/core/src/test/java/test/org/apache/spark/sql/execution/sort/RecordBinaryComparatorSuite.java
index 564e76737ecde..6cb7c40e3332b 100644
--- a/sql/core/src/test/java/test/org/apache/spark/sql/execution/sort/RecordBinaryComparatorSuite.java
+++ b/sql/core/src/test/java/test/org/apache/spark/sql/execution/sort/RecordBinaryComparatorSuite.java
@@ -37,6 +37,8 @@
 import org.junit.Before;
 import org.junit.Test;
 
+import java.nio.ByteOrder;
+
 /**
  * Test the RecordBinaryComparator, which compares two UnsafeRows by their binary form.
  */
@@ -261,40 +263,74 @@ public void testBinaryComparatorForNullColumns() throws Exception {
   public void testBinaryComparatorWhenSubtractionIsDivisibleByMaxIntValue() throws Exception {
     int numFields = 1;
 
+    // Place the following bytes (hex) into UnsafeRows for the comparison:
+    //
+    //   index | 00 01 02 03 04 05 06 07
+    //   ------+------------------------
+    //   row1  | 00 00 00 00 00 00 00 0b
+    //   row2  | 00 00 00 00 80 00 00 0a
+    //
+    // The byte layout needs to be identical on all platforms regardless of
+    // of endianness. To achieve this the bytes in each value are reversed
+    // on little-endian platforms.
+    long row1Data = 11L;
+    long row2Data = 11L + Integer.MAX_VALUE;
+    if (ByteOrder.nativeOrder().equals(ByteOrder.LITTLE_ENDIAN)) {
+      row1Data = Long.reverseBytes(row1Data);
+      row2Data = Long.reverseBytes(row2Data);
+    }
+
     UnsafeRow row1 = new UnsafeRow(numFields);
     byte[] data1 = new byte[100];
     row1.pointTo(data1, computeSizeInBytes(numFields * 8));
-    row1.setLong(0, 11);
+    row1.setLong(0, row1Data);
 
     UnsafeRow row2 = new UnsafeRow(numFields);
     byte[] data2 = new byte[100];
     row2.pointTo(data2, computeSizeInBytes(numFields * 8));
-    row2.setLong(0, 11L + Integer.MAX_VALUE);
+    row2.setLong(0, row2Data);
 
     insertRow(row1);
     insertRow(row2);
 
-    Assert.assertTrue(compare(0, 1) > 0);
+    Assert.assertTrue(compare(0, 1) < 0);
   }
 
   @Test
   public void testBinaryComparatorWhenSubtractionCanOverflowLongValue() throws Exception {
     int numFields = 1;
 
+    // Place the following bytes (hex) into UnsafeRows for the comparison:
+    //
+    //   index | 00 01 02 03 04 05 06 07
+    //   ------+------------------------
+    //   row1  | 80 00 00 00 00 00 00 00
+    //   row2  | 00 00 00 00 00 00 00 01
+    //
+    // The byte layout needs to be identical on all platforms regardless of
+    // of endianness. To achieve this the bytes in each value are reversed
+    // on little-endian platforms.
+    long row1Data = Long.MIN_VALUE;
+    long row2Data = 1L;
+    if (ByteOrder.nativeOrder().equals(ByteOrder.LITTLE_ENDIAN)) {
+      row1Data = Long.reverseBytes(row1Data);
+      row2Data = Long.reverseBytes(row2Data);
+    }
+
     UnsafeRow row1 = new UnsafeRow(numFields);
     byte[] data1 = new byte[100];
     row1.pointTo(data1, computeSizeInBytes(numFields * 8));
-    row1.setLong(0, Long.MIN_VALUE);
+    row1.setLong(0, row1Data);
 
     UnsafeRow row2 = new UnsafeRow(numFields);
     byte[] data2 = new byte[100];
     row2.pointTo(data2, computeSizeInBytes(numFields * 8));
-    row2.setLong(0, 1);
+    row2.setLong(0, row2Data);
 
     insertRow(row1);
     insertRow(row2);
 
-    Assert.assertTrue(compare(0, 1) < 0);
+    Assert.assertTrue(compare(0, 1) > 0);
   }
 
   @Test