[SPARK-59430][CORE] Cache UTF8String.hashCode

common/unsafe/src/main/java/org/apache/spark/unsafe/types/UTF8String.java

@@ -62,6 +62,12 @@ public final class UTF8String implements Comparable<UTF8String>, Externalizable,
   private int numBytes;
   private volatile int numChars = -1;
 
+  // Cache the hash code, like java.lang.String. `hash` holds the computed value; `hashIsZero`
+  // records the case where the hash legitimately computed to 0, so it is not recomputed on every
+  // call. Non-volatile: the benign race only ever recomputes the same value.
+  private int hash;
+  private boolean hashIsZero;
+
   /**
    * The validity of the UTF8Strings can be cached to avoid repeated validation checks, because
    * that operation requires full string scan. Valid strings have no illegal UTF-8 byte sequences.
@@ -2276,7 +2282,16 @@ public int levenshteinDistance(UTF8String other, int threshold) {
 
   @Override
   public int hashCode() {
-    return Murmur3_x86_32.hashUnsafeBytes(base, offset, numBytes, 42);
+    int h = hash;
+    if (h == 0 && !hashIsZero) {
+      h = Murmur3_x86_32.hashUnsafeBytes(base, offset, numBytes, 42);
+      if (h == 0) {
+        hashIsZero = true;
+      } else {
+        hash = h;
+      }
+    }
+    return h;
   }
 
   /**
@@ -2345,6 +2360,8 @@ public void readExternal(ObjectInput in) throws IOException, ClassNotFoundExcept
     numBytes = in.readInt();
     base = new byte[numBytes];
     in.readFully((byte[]) base);
+    hash = 0;
+    hashIsZero = false;
   }
 
   @Override
@@ -2360,6 +2377,8 @@ public void read(Kryo kryo, Input in) {
     this.numBytes = in.readInt();
     this.base = new byte[numBytes];
     in.read((byte[]) base);
+    this.hash = 0;
+    this.hashIsZero = false;
   }
 
   /**

common/unsafe/src/test/java/org/apache/spark/unsafe/types/UTF8StringSuite.java

@@ -17,8 +17,11 @@
 
 package org.apache.spark.unsafe.types;
 
+import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
 import java.nio.charset.StandardCharsets;
@@ -1548,4 +1551,31 @@ public void testAppendCodepointToUTF8StringBuilder() {
       assert(usb.build().equals(UTF8String.fromString(sb.toString())));
     }
   }
+
+  @Test
+  public void hashCodeIsCachedAndStable() throws Exception {
+    // Repeated calls return the same value (the cache must not change it).
+    UTF8String s = fromString("hashcode-cache-test");
+    int h = s.hashCode();
+    assertEquals(h, s.hashCode());
+    assertEquals(h, s.hashCode());
+    // Equal strings built differently agree (consistency with equals).
+    assertEquals(h, fromBytes("hashcode-cache-test".getBytes(StandardCharsets.UTF_8)).hashCode());
+
+    // A Java-serialization round-trip (which re-points the fields via readExternal) yields the
+    // same hash, even when the source's hash was already cached.
+    UTF8String before = fromString("round-trip");
+    before.hashCode();
+    ByteArrayOutputStream bos = new ByteArrayOutputStream();
+    try (ObjectOutputStream oos = new ObjectOutputStream(bos)) {
+      oos.writeObject(before);
+    }
+    ByteArrayInputStream bis = new ByteArrayInputStream(bos.toByteArray());
+    UTF8String after;
+    try (ObjectInputStream ois = new ObjectInputStream(bis)) {
+      after = (UTF8String) ois.readObject();
+    }
+    assertEquals(before, after);
+    assertEquals(before.hashCode(), after.hashCode());
+  }
 }