[3.2][Kernel] Change string comparing from UTF16 to UTF8 (#3611)

Changed string comparing from UTF16 to UTF8. This fixes comparison
issues around the characters with surrogate pairs.

Tests added to `DefaultExpressionEvaluatorSuite.scala`

Author: ilicmarkodb

kernel/kernel-defaults/src/main/java/io/delta/kernel/defaults/internal/expressions/DefaultExpressionUtils.java

@@ -16,6 +16,7 @@
 package io.delta.kernel.defaults.internal.expressions;
 
 import java.math.BigDecimal;
+import java.nio.charset.StandardCharsets;
 import java.util.Comparator;
 import java.util.List;
 import java.util.function.Function;
@@ -35,6 +36,19 @@
 class DefaultExpressionUtils {
     private DefaultExpressionUtils() {}
 
+    static final Comparator<String> STRING_COMPARATOR = (leftOp, rightOp) -> {
+        byte[] leftBytes = leftOp.getBytes(StandardCharsets.UTF_8);
+        byte[] rightBytes = rightOp.getBytes(StandardCharsets.UTF_8);
+        int i = 0;
+        while (i < leftBytes.length && i < rightBytes.length) {
+            if (leftBytes[i] != rightBytes[i]) {
+                return Byte.toUnsignedInt(leftBytes[i]) - Byte.toUnsignedInt(rightBytes[i]);
+            }
+            i++;
+        }
+        return Integer.compare(leftBytes.length, rightBytes.length);
+    };
+
     /**
      * Utility method that calculates the nullability result from given two vectors. Result is
      * null if at least one side is a null.
@@ -186,10 +200,10 @@ static void compareDouble(ColumnVector left, ColumnVector right, int[] result) {
     }
 
     static void compareString(ColumnVector left, ColumnVector right, int[] result) {
-        Comparator<String> comparator = Comparator.naturalOrder();
         for (int rowId = 0; rowId < left.getSize(); rowId++) {
             if (!left.isNullAt(rowId) && !right.isNullAt(rowId)) {
-                result[rowId] = comparator.compare(left.getString(rowId), right.getString(rowId));
+                result[rowId] =
+                        STRING_COMPARATOR.compare(left.getString(rowId), right.getString(rowId));
             }
         }
     }

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/internal/expressions/DefaultExpressionEvaluatorSuite.scala

@@ -308,6 +308,9 @@ class DefaultExpressionEvaluatorSuite extends AnyFunSuite with ExpressionSuiteBa
   }
 
   test("evaluate expression: comparators (=, <, <=, >, >=)") {
+    val ASCII_MAX_CHARACTER = '\u007F'
+    val UTF8_MAX_CHARACTER = new String(Character.toChars(Character.MAX_CODE_POINT))
+
     // Literals for each data type from the data type value range, used as inputs to comparator
     // (small, big, small, null)
     val literals = Seq(
@@ -332,6 +335,66 @@ class DefaultExpressionEvaluatorSuite extends AnyFunSuite with ExpressionSuiteBa
       ),
       (ofDate(-12123), ofDate(123123), ofDate(-12123), ofNull(DateType.DATE)),
       (ofString("apples"), ofString("oranges"), ofString("apples"), ofNull(StringType.STRING)),
+      (ofString(""), ofString("a"), ofString(""), ofNull(StringType.STRING)),
+      (ofString("abc"), ofString("abc0"), ofString("abc"), ofNull(StringType.STRING)),
+      (ofString("abc"), ofString("abcd"), ofString("abc"), ofNull(StringType.STRING)),
+      (ofString("abc"), ofString("abd"), ofString("abc"), ofNull(StringType.STRING)),
+      (
+        ofString("Abcabcabc"),
+        ofString("aBcabcabc"),
+        ofString("Abcabcabc"),
+        ofNull(StringType.STRING)
+      ),
+      (
+        ofString("abcabcabC"),
+        ofString("abcabcabc"),
+        ofString("abcabcabC"),
+        ofNull(StringType.STRING)
+      ),
+      // scalastyle:off nonascii
+      (ofString("abc"), ofString("世界"), ofString("abc"), ofNull(StringType.STRING)),
+      (ofString("世界"), ofString("你好"), ofString("世界"), ofNull(StringType.STRING)),
+      (ofString("你好122"), ofString("你好123"), ofString("你好122"), ofNull(StringType.STRING)),
+      (ofString("A"), ofString("Ā"), ofString("A"), ofNull(StringType.STRING)),
+      (ofString("»"), ofString("î"), ofString("»"), ofNull(StringType.STRING)),
+      (ofString("�"), ofString("🌼"), ofString("�"), ofNull(StringType.STRING)),
+      (
+        ofString("abcdef🚀"),
+        ofString(s"abcdef$UTF8_MAX_CHARACTER"),
+        ofString("abcdef🚀"),
+        ofNull(StringType.STRING)
+      ),
+      (
+        ofString("abcde�abcdef�abcdef�abcdef"),
+        ofString(s"abcde�$ASCII_MAX_CHARACTER"),
+        ofString("abcde�abcdef�abcdef�abcdef"),
+        ofNull(StringType.STRING)
+      ),
+      (
+        ofString("abcde�abcdef�abcdef�abcdef"),
+        ofString(s"abcde�$ASCII_MAX_CHARACTER"),
+        ofString("abcde�abcdef�abcdef�abcdef"),
+        ofNull(StringType.STRING)
+      ),
+      (
+        ofString("����"),
+        ofString(s"��$UTF8_MAX_CHARACTER"),
+        ofString("����"),
+        ofNull(StringType.STRING)
+      ),
+      (
+        ofString(s"a${UTF8_MAX_CHARACTER}d"),
+        ofString(s"a$UTF8_MAX_CHARACTER$ASCII_MAX_CHARACTER"),
+        ofString(s"a${UTF8_MAX_CHARACTER}d"),
+        ofNull(StringType.STRING)
+      ),
+      (
+        ofString("abcdefghijklm💞😉💕\n🥀🌹💐🌺🌷🌼🌻🌷🥀"),
+        ofString(s"abcdefghijklm💞😉💕\n🥀🌹💐🌺🌷🌼$UTF8_MAX_CHARACTER"),
+        ofString("abcdefghijklm💞😉💕\n🥀🌹💐🌺🌷🌼🌻🌷🥀"),
+        ofNull(StringType.STRING)
+      ),
+      // scalastyle:on nonascii
       (
         ofBinary("apples".getBytes()),
         ofBinary("oranges".getBytes()),