diff --git a/core/src/main/java/org/apache/spark/util/collection/unsafe/sort/PrefixComparators.java b/core/src/main/java/org/apache/spark/util/collection/unsafe/sort/PrefixComparators.java
index 5f82579c0267b..d5deda76c84eb 100644
--- a/core/src/main/java/org/apache/spark/util/collection/unsafe/sort/PrefixComparators.java
+++ b/core/src/main/java/org/apache/spark/util/collection/unsafe/sort/PrefixComparators.java
@@ -19,6 +19,7 @@
 
 import com.google.common.base.Charsets;
 import com.google.common.primitives.Longs;
+
 import org.apache.spark.annotation.Private;
 import org.apache.spark.unsafe.types.UTF8String;
 
@@ -27,8 +28,7 @@ public class PrefixComparators {
   private PrefixComparators() {}
 
   public static final StringPrefixComparator STRING = new StringPrefixComparator();
-  public static final IntPrefixComparator INTEGER = new IntPrefixComparator();
-  public static final LongPrefixComparator LONG = new LongPrefixComparator();
+  public static final IntegralPrefixComparator INTEGRAL = new IntegralPrefixComparator();
   public static final FloatPrefixComparator FLOAT = new FloatPrefixComparator();
   public static final DoublePrefixComparator DOUBLE = new DoublePrefixComparator();
 
@@ -67,22 +67,10 @@ public long computePrefix(String value) {
     }
   }
 
-  public static final class IntPrefixComparator extends PrefixComparator {
-    @Override
-    public int compare(long aPrefix, long bPrefix) {
-      int a = (int) aPrefix;
-      int b = (int) bPrefix;
-      return (a < b) ? -1 : (a > b) ? 1 : 0;
-    }
-
-    public final long NULL_PREFIX = computePrefix(Integer.MIN_VALUE);
-
-    public long computePrefix(int value) {
-      return value & 0xffffffffL;
-    }
-  }
-
-  public static final class LongPrefixComparator extends PrefixComparator {
+  /**
+   * Prefix comparator for all integral types (boolean, byte, short, int, long).
+   */
+  public static final class IntegralPrefixComparator extends PrefixComparator {
     @Override
     public int compare(long a, long b) {
       return (a < b) ? -1 : (a > b) ? 1 : 0;
diff --git a/core/src/test/java/org/apache/spark/util/collection/unsafe/sort/PrefixComparatorsSuite.java b/core/src/test/java/org/apache/spark/util/collection/unsafe/sort/PrefixComparatorsSuite.java
deleted file mode 100644
index 4c4a7d5f5486a..0000000000000
--- a/core/src/test/java/org/apache/spark/util/collection/unsafe/sort/PrefixComparatorsSuite.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.spark.util.collection.unsafe.sort;
-
-import org.junit.Test;
-import static org.junit.Assert.*;
-
-public class PrefixComparatorsSuite {
-
-  private static int genericComparison(Comparable a, Comparable b) {
-    return a.compareTo(b);
-  }
-
-  @Test
-  public void intPrefixComparator() {
-    int[] testData = new int[] { 0, Integer.MIN_VALUE, Integer.MAX_VALUE, 0, 1, 2, -1, -2, 1024};
-    for (int a : testData) {
-      for (int b : testData) {
-        long aPrefix = PrefixComparators.INTEGER.computePrefix(a);
-        long bPrefix = PrefixComparators.INTEGER.computePrefix(b);
-        assertEquals(
-          "Wrong prefix comparison results for a=" + a + " b=" + b,
-          genericComparison(a, b),
-          PrefixComparators.INTEGER.compare(aPrefix, bPrefix));
-
-      }
-    }
-  }
-
-}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/SortPrefixUtils.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/SortPrefixUtils.scala
index 366b09ddfbfdb..2dee3542d6101 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/SortPrefixUtils.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/SortPrefixUtils.scala
@@ -38,8 +38,7 @@ object SortPrefixUtils {
   def getPrefixComparator(sortOrder: SortOrder): PrefixComparator = {
     sortOrder.dataType match {
       case StringType => PrefixComparators.STRING
-      case IntegerType => PrefixComparators.INTEGER
-      case LongType => PrefixComparators.LONG
+      case BooleanType | ByteType | ShortType | IntegerType | LongType => PrefixComparators.INTEGRAL
       case FloatType => PrefixComparators.FLOAT
       case DoubleType => PrefixComparators.DOUBLE
       case _ => NoOpPrefixComparator
@@ -51,16 +50,37 @@ object SortPrefixUtils {
       case StringType => (row: InternalRow) => {
         PrefixComparators.STRING.computePrefix(sortOrder.child.eval(row).asInstanceOf[UTF8String])
       }
-      case IntegerType => (row: InternalRow) => {
-        val exprVal = sortOrder.child.eval(row)
-        if (exprVal == null) PrefixComparators.INTEGER.NULL_PREFIX
-        else PrefixComparators.INTEGER.computePrefix(sortOrder.child.eval(row).asInstanceOf[Int])
-      }
-      case LongType => (row: InternalRow) => {
-        val exprVal = sortOrder.child.eval(row)
-        if (exprVal == null) PrefixComparators.LONG.NULL_PREFIX
-        else sortOrder.child.eval(row).asInstanceOf[Long]
-      }
+      case BooleanType =>
+        (row: InternalRow) => {
+          val exprVal = sortOrder.child.eval(row)
+          if (exprVal == null) PrefixComparators.INTEGRAL.NULL_PREFIX
+          else if (sortOrder.child.eval(row).asInstanceOf[Boolean]) 1
+          else 0
+        }
+      case ByteType =>
+        (row: InternalRow) => {
+          val exprVal = sortOrder.child.eval(row)
+          if (exprVal == null) PrefixComparators.INTEGRAL.NULL_PREFIX
+          else sortOrder.child.eval(row).asInstanceOf[Byte]
+        }
+      case ShortType =>
+        (row: InternalRow) => {
+          val exprVal = sortOrder.child.eval(row)
+          if (exprVal == null) PrefixComparators.INTEGRAL.NULL_PREFIX
+          else sortOrder.child.eval(row).asInstanceOf[Short]
+        }
+      case IntegerType =>
+        (row: InternalRow) => {
+          val exprVal = sortOrder.child.eval(row)
+          if (exprVal == null) PrefixComparators.INTEGRAL.NULL_PREFIX
+          else sortOrder.child.eval(row).asInstanceOf[Int]
+        }
+      case LongType =>
+        (row: InternalRow) => {
+          val exprVal = sortOrder.child.eval(row)
+          if (exprVal == null) PrefixComparators.INTEGRAL.NULL_PREFIX
+          else sortOrder.child.eval(row).asInstanceOf[Long]
+        }
       case FloatType => (row: InternalRow) => {
         val exprVal = sortOrder.child.eval(row)
         if (exprVal == null) PrefixComparators.FLOAT.NULL_PREFIX