diff --git a/common/kvstore/src/main/java/org/apache/spark/util/kvstore/LevelDBTypeInfo.java b/common/kvstore/src/main/java/org/apache/spark/util/kvstore/LevelDBTypeInfo.java
index 4d7f76f673865..d313dcd6c779c 100644
--- a/common/kvstore/src/main/java/org/apache/spark/util/kvstore/LevelDBTypeInfo.java
+++ b/common/kvstore/src/main/java/org/apache/spark/util/kvstore/LevelDBTypeInfo.java
@@ -461,6 +461,15 @@ byte[] toKey(Object value, byte prefix) {
 
       if (value instanceof String) {
         byte[] str = ((String) value).getBytes(UTF_8);
+        for (byte b : str) {
+          if (b == KEY_SEPARATOR) {
+            // String bytes are copied verbatim into the key, so an embedded key
+            // separator would corrupt the index key structure and mix entries across
+            // range scans and deletes. Reject such values outright.
+            throw new IllegalArgumentException(
+              "Index string values cannot contain the key separator (0x00) byte.");
+          }
+        }
         result = new byte[str.length + 1];
         result[0] = prefix;
         System.arraycopy(str, 0, result, 1, str.length);
diff --git a/common/kvstore/src/main/java/org/apache/spark/util/kvstore/RocksDBTypeInfo.java b/common/kvstore/src/main/java/org/apache/spark/util/kvstore/RocksDBTypeInfo.java
index c5452b4104ba2..525051257dc83 100644
--- a/common/kvstore/src/main/java/org/apache/spark/util/kvstore/RocksDBTypeInfo.java
+++ b/common/kvstore/src/main/java/org/apache/spark/util/kvstore/RocksDBTypeInfo.java
@@ -463,6 +463,15 @@ byte[] toKey(Object value, byte prefix) {
 
       if (value instanceof String) {
         byte[] str = ((String) value).getBytes(UTF_8);
+        for (byte b : str) {
+          if (b == KEY_SEPARATOR) {
+            // String bytes are copied verbatim into the key, so an embedded key
+            // separator would corrupt the index key structure and mix entries across
+            // range scans and deletes. Reject such values outright.
+            throw new IllegalArgumentException(
+              "Index string values cannot contain the key separator (0x00) byte.");
+          }
+        }
         result = new byte[str.length + 1];
         result[0] = prefix;
         System.arraycopy(str, 0, result, 1, str.length);
diff --git a/common/kvstore/src/test/java/org/apache/spark/util/kvstore/LevelDBTypeInfoSuite.java b/common/kvstore/src/test/java/org/apache/spark/util/kvstore/LevelDBTypeInfoSuite.java
index 0359e11404cd4..ae269076f4ce3 100644
--- a/common/kvstore/src/test/java/org/apache/spark/util/kvstore/LevelDBTypeInfoSuite.java
+++ b/common/kvstore/src/test/java/org/apache/spark/util/kvstore/LevelDBTypeInfoSuite.java
@@ -210,4 +210,12 @@ public String id(boolean illegalParam) {
 
   }
 
+
+  @Test
+  public void testStringValuesRejectKeySeparator() throws Exception {
+    LevelDBTypeInfo.Index idx = newTypeInfo(CustomType1.class).indices().iterator().next();
+    assertThrows(IllegalArgumentException.class, () -> idx.toKey("key\u0000suffix"));
+    assertThrows(IllegalArgumentException.class,
+      () -> idx.toKey(new String[] { "ok", "bad\u0000" }));
+  }
 }
diff --git a/common/kvstore/src/test/java/org/apache/spark/util/kvstore/RocksDBTypeInfoSuite.java b/common/kvstore/src/test/java/org/apache/spark/util/kvstore/RocksDBTypeInfoSuite.java
index f694fd36b68b3..5b606cc601459 100644
--- a/common/kvstore/src/test/java/org/apache/spark/util/kvstore/RocksDBTypeInfoSuite.java
+++ b/common/kvstore/src/test/java/org/apache/spark/util/kvstore/RocksDBTypeInfoSuite.java
@@ -210,4 +210,12 @@ public String id(boolean illegalParam) {
 
   }
 
+
+  @Test
+  public void testStringValuesRejectKeySeparator() throws Exception {
+    RocksDBTypeInfo.Index idx = newTypeInfo(CustomType1.class).indices().iterator().next();
+    assertThrows(IllegalArgumentException.class, () -> idx.toKey("key\u0000suffix"));
+    assertThrows(IllegalArgumentException.class,
+      () -> idx.toKey(new String[] { "ok", "bad\u0000" }));
+  }
 }