diff --git a/common/kvstore/src/main/java/org/apache/spark/util/kvstore/KVStore.java b/common/kvstore/src/main/java/org/apache/spark/util/kvstore/KVStore.java
index abadbfb4d997b..959e5b1fd1ec6 100644
--- a/common/kvstore/src/main/java/org/apache/spark/util/kvstore/KVStore.java
+++ b/common/kvstore/src/main/java/org/apache/spark/util/kvstore/KVStore.java
@@ -104,11 +104,11 @@ public interface KVStore extends Closeable {
 
   /**
    * Removes an object and all data related to it, like index entries, from the store.
+   * Deleting a missing key is a no-op.
    *
    * @param type The object's type.
    * @param naturalKey The object's "natural key", which uniquely identifies it. Null keys
    *                   are not allowed.
-   * @throws java.util.NoSuchElementException If an element with the given key does not exist.
    */
   void delete(Class<?> type, Object naturalKey) throws Exception;
 
diff --git a/common/kvstore/src/main/java/org/apache/spark/util/kvstore/LevelDB.java b/common/kvstore/src/main/java/org/apache/spark/util/kvstore/LevelDB.java
index 91b2cde2d84fe..57d26d09b52ad 100644
--- a/common/kvstore/src/main/java/org/apache/spark/util/kvstore/LevelDB.java
+++ b/common/kvstore/src/main/java/org/apache/spark/util/kvstore/LevelDB.java
@@ -98,12 +98,8 @@ public LevelDB(File path, KVStoreSerializer serializer) throws Exception {
       db().put(STORE_VERSION_KEY, serializer.serialize(STORE_VERSION));
     }
 
-    Map<String, byte[]> aliases;
-    try {
-      aliases = get(TYPE_ALIASES_KEY, TypeAliases.class).aliases;
-    } catch (NoSuchElementException e) {
-      aliases = new HashMap<>();
-    }
+    TypeAliases aliasesValue = getOrNull(TYPE_ALIASES_KEY, TypeAliases.class);
+    Map<String, byte[]> aliases = aliasesValue != null ? aliasesValue.aliases : new HashMap<>();
     typeAliases = new ConcurrentHashMap<>(aliases);
 
     iteratorTracker = new ConcurrentLinkedQueue<>();
@@ -111,11 +107,7 @@ public LevelDB(File path, KVStoreSerializer serializer) throws Exception {
 
   @Override
   public <T> T getMetadata(Class<T> klass) throws Exception {
-    try {
-      return get(METADATA_KEY, klass);
-    } catch (NoSuchElementException nsee) {
-      return null;
-    }
+    return getOrNull(METADATA_KEY, klass);
   }
 
   @Override
@@ -128,11 +120,21 @@ public void setMetadata(Object value) throws Exception {
   }
 
   <T> T get(byte[] key, Class<T> klass) throws Exception {
-    byte[] data = db().get(key);
-    if (data == null) {
+    T value = getOrNull(key, klass);
+    if (value == null) {
       throw new NoSuchElementException(new String(key, UTF_8));
     }
-    return serializer.deserialize(data, klass);
+    return value;
+  }
+
+  /**
+   * Returns the value for the given key, or {@code null} if the key is not present, so that
+   * callers where a missing key is expected do not pay the cost of throwing and filling in an
+   * exception stack trace.
+   */
+  <T> T getOrNull(byte[] key, Class<T> klass) throws Exception {
+    byte[] data = db().get(key);
+    return data != null ? serializer.deserialize(data, klass) : null;
   }
 
   private void put(byte[] key, Object value) throws Exception {
@@ -207,12 +209,7 @@ private void updateBatch(
       Class<?> klass,
       LevelDBTypeInfo.Index naturalIndex,
       Collection<LevelDBTypeInfo.Index> indices) throws Exception {
-    Object existing;
-    try {
-      existing = get(naturalIndex.entityKey(null, value), klass);
-    } catch (NoSuchElementException e) {
-      existing = null;
-    }
+    Object existing = getOrNull(naturalIndex.entityKey(null, value), klass);
 
     PrefixCache cache = new PrefixCache(value);
     byte[] naturalKey = naturalIndex.toKey(naturalIndex.getValue(value));
@@ -229,9 +226,8 @@ public void delete(Class<?> type, Object naturalKey) throws Exception {
       LevelDBTypeInfo ti = getTypeInfo(type);
       byte[] key = ti.naturalIndex().start(null, naturalKey);
       synchronized (ti) {
-        byte[] data = db().get(key);
-        if (data != null) {
-          Object existing = serializer.deserialize(data, type);
+        Object existing = getOrNull(key, type);
+        if (existing != null) {
           PrefixCache cache = new PrefixCache(existing);
           byte[] keyBytes = ti.naturalIndex().toKey(ti.naturalIndex().getValue(existing));
           for (LevelDBTypeInfo.Index idx : ti.indices()) {
@@ -240,8 +236,6 @@ public void delete(Class<?> type, Object naturalKey) throws Exception {
           db().write(batch);
         }
       }
-    } catch (NoSuchElementException nse) {
-      // Ignore.
     }
   }
 
diff --git a/common/kvstore/src/main/java/org/apache/spark/util/kvstore/RocksDB.java b/common/kvstore/src/main/java/org/apache/spark/util/kvstore/RocksDB.java
index 4b69b9441dc32..cd4e18a68a2ee 100644
--- a/common/kvstore/src/main/java/org/apache/spark/util/kvstore/RocksDB.java
+++ b/common/kvstore/src/main/java/org/apache/spark/util/kvstore/RocksDB.java
@@ -131,12 +131,8 @@ public RocksDB(File path, KVStoreSerializer serializer) throws Exception {
       db().put(STORE_VERSION_KEY, serializer.serialize(STORE_VERSION));
     }
 
-    Map<String, byte[]> aliases;
-    try {
-      aliases = get(TYPE_ALIASES_KEY, TypeAliases.class).aliases;
-    } catch (NoSuchElementException e) {
-      aliases = new HashMap<>();
-    }
+    TypeAliases aliasesValue = getOrNull(TYPE_ALIASES_KEY, TypeAliases.class);
+    Map<String, byte[]> aliases = aliasesValue != null ? aliasesValue.aliases : new HashMap<>();
     typeAliases = new ConcurrentHashMap<>(aliases);
 
     iteratorTracker = new ConcurrentLinkedQueue<>();
@@ -144,11 +140,7 @@ public RocksDB(File path, KVStoreSerializer serializer) throws Exception {
 
   @Override
   public <T> T getMetadata(Class<T> klass) throws Exception {
-    try {
-      return get(METADATA_KEY, klass);
-    } catch (NoSuchElementException nsee) {
-      return null;
-    }
+    return getOrNull(METADATA_KEY, klass);
   }
 
   @Override
@@ -161,11 +153,21 @@ public void setMetadata(Object value) throws Exception {
   }
 
   <T> T get(byte[] key, Class<T> klass) throws Exception {
-    byte[] data = db().get(key);
-    if (data == null) {
+    T value = getOrNull(key, klass);
+    if (value == null) {
       throw new NoSuchElementException(new String(key, UTF_8));
     }
-    return serializer.deserialize(data, klass);
+    return value;
+  }
+
+  /**
+   * Returns the value for the given key, or {@code null} if the key is not present, so that
+   * callers where a missing key is expected do not pay the cost of throwing and filling in an
+   * exception stack trace.
+   */
+  <T> T getOrNull(byte[] key, Class<T> klass) throws Exception {
+    byte[] data = db().get(key);
+    return data != null ? serializer.deserialize(data, klass) : null;
   }
 
   private void put(byte[] key, Object value) throws Exception {
@@ -239,12 +241,7 @@ private void updateBatch(
       Class<?> klass,
       RocksDBTypeInfo.Index naturalIndex,
       Collection<RocksDBTypeInfo.Index> indices) throws Exception {
-    Object existing;
-    try {
-      existing = get(naturalIndex.entityKey(null, value), klass);
-    } catch (NoSuchElementException e) {
-      existing = null;
-    }
+    Object existing = getOrNull(naturalIndex.entityKey(null, value), klass);
 
     PrefixCache cache = new PrefixCache(value);
     byte[] naturalKey = naturalIndex.toKey(naturalIndex.getValue(value));
@@ -261,9 +258,8 @@ public void delete(Class<?> type, Object naturalKey) throws Exception {
       RocksDBTypeInfo ti = getTypeInfo(type);
       byte[] key = ti.naturalIndex().start(null, naturalKey);
       synchronized (ti) {
-        byte[] data = db().get(key);
-        if (data != null) {
-          Object existing = serializer.deserialize(data, type);
+        Object existing = getOrNull(key, type);
+        if (existing != null) {
           PrefixCache cache = new PrefixCache(existing);
           byte[] keyBytes = ti.naturalIndex().toKey(ti.naturalIndex().getValue(existing));
           for (RocksDBTypeInfo.Index idx : ti.indices()) {
@@ -272,8 +268,6 @@ public void delete(Class<?> type, Object naturalKey) throws Exception {
           db().write(writeOptions, writeBatch);
         }
       }
-    } catch (NoSuchElementException nse) {
-      // Ignore.
     }
   }
 
diff --git a/common/kvstore/src/test/java/org/apache/spark/util/kvstore/LevelDBSuite.java b/common/kvstore/src/test/java/org/apache/spark/util/kvstore/LevelDBSuite.java
index 25e6664d28dd1..79db74c383993 100644
--- a/common/kvstore/src/test/java/org/apache/spark/util/kvstore/LevelDBSuite.java
+++ b/common/kvstore/src/test/java/org/apache/spark/util/kvstore/LevelDBSuite.java
@@ -98,6 +98,32 @@ public void testObjectWriteReadDelete() throws Exception {
     assertEquals(0, countKeys(t.getClass()));
   }
 
+  @Test
+  public void testGetOrNullMissingKey() throws Exception {
+    // getOrNull() returns null for a missing key so expected misses (e.g. the write path
+    // looking up an existing entry) skip the cost of building an exception, while read()
+    // still surfaces a missing key as NoSuchElementException.
+    byte[] missingKey = db.getTypeInfo(CustomType1.class).naturalIndex().start(null, "missing");
+    assertNull(db.getOrNull(missingKey, CustomType1.class));
+    assertThrows(NoSuchElementException.class, () -> db.read(CustomType1.class, "missing"));
+
+    CustomType1 t = createCustomType1(1);
+    db.write(t);
+    assertEquals(t, db.read(CustomType1.class, t.key));
+  }
+
+  @Test
+  public void testDeleteEdgeCases() throws Exception {
+    // Never-written type: type info is created on the fly, lookup misses, nothing happens.
+    db.delete(CustomType1.class, "missing");
+    assertEquals(0L, db.count(CustomType1.class));
+
+    // Never-written key of a written type.
+    db.write(createCustomType1(1));
+    db.delete(CustomType1.class, "missing");
+    assertEquals(1L, db.count(CustomType1.class));
+  }
+
   @Test
   public void testMultipleObjectWriteReadDelete() throws Exception {
     CustomType1 t1 = createCustomType1(1);
diff --git a/common/kvstore/src/test/java/org/apache/spark/util/kvstore/RocksDBSuite.java b/common/kvstore/src/test/java/org/apache/spark/util/kvstore/RocksDBSuite.java
index 1334386fde74d..8de0ea3d8576e 100644
--- a/common/kvstore/src/test/java/org/apache/spark/util/kvstore/RocksDBSuite.java
+++ b/common/kvstore/src/test/java/org/apache/spark/util/kvstore/RocksDBSuite.java
@@ -95,6 +95,32 @@ public void testObjectWriteReadDelete() throws Exception {
     assertEquals(0, countKeys(t.getClass()));
   }
 
+  @Test
+  public void testGetOrNullMissingKey() throws Exception {
+    // getOrNull() returns null for a missing key so expected misses (e.g. the write path
+    // looking up an existing entry) skip the cost of building an exception, while read()
+    // still surfaces a missing key as NoSuchElementException.
+    byte[] missingKey = db.getTypeInfo(CustomType1.class).naturalIndex().start(null, "missing");
+    assertNull(db.getOrNull(missingKey, CustomType1.class));
+    assertThrows(NoSuchElementException.class, () -> db.read(CustomType1.class, "missing"));
+
+    CustomType1 t = createCustomType1(1);
+    db.write(t);
+    assertEquals(t, db.read(CustomType1.class, t.key));
+  }
+
+  @Test
+  public void testDeleteEdgeCases() throws Exception {
+    // Never-written type: type info is created on the fly, lookup misses, nothing happens.
+    db.delete(CustomType1.class, "missing");
+    assertEquals(0L, db.count(CustomType1.class));
+
+    // Never-written key of a written type.
+    db.write(createCustomType1(1));
+    db.delete(CustomType1.class, "missing");
+    assertEquals(1L, db.count(CustomType1.class));
+  }
+
   @Test
   public void testMultipleObjectWriteReadDelete() throws Exception {
     CustomType1 t1 = createCustomType1(1);