diff --git a/docs/commands.md b/docs/commands.md
index c84367e60..b87264a24 100644
--- a/docs/commands.md
+++ b/docs/commands.md
@@ -95,7 +95,7 @@ Purge old block data. Useful for freeing up space on your HDD if you don't need
 
 | Command | Parameters |
 | --- | --- |
-| /co purge | `t:<time> r:<world> i:<include>` |
+| /co purge | `t:<time> r:<world> i:<include> e:<exclude> a:kill` |
 
 For example, `/co purge t:30d` will delete all data older than one month, and only keep the last 30 days of data.
 
@@ -110,12 +110,22 @@ For example, `/co purge t:30d r:#world_nether` will delete all data older than o
 You can optionally specify block types in CoreProtect v23+.  
 For example, `/co purge t:30d i:stone,dirt` will delete all stone and dirt data older than one month, without removing other block data.
 
+**Purging Entity Kills**  
+Entity kill logs can be purged separately from other data on SQLite, MySQL, and DuckDB. Each kill also removes the saved entity data used to restore the mob on rollback.
+
+* `/co purge t:30d a:kill` deletes all entity kills older than one month, and keeps all other data.
+* `/co purge t:30d i:zombie,skeleton` deletes zombie and skeleton kills older than one month. Entity and block types can be combined in `i:`.
+* `/co purge t:30d e:villager,wolf` deletes all data older than one month, but keeps villager and wolf kills.
+* `/co purge t:30d a:kill e:villager r:#world` deletes entity kills older than one month in the overworld, except villager kills.
+
+`e:` only accepts entity types. It cannot be combined with entity types in `i:`, or with only block types in `i:`, because a block restriction already keeps every kill. `a:kill` cannot be combined with block types in `i:`.
+
 **Database Optimization**
 
 In CoreProtect v2.15+, adding `#optimize` to the end of the command (for example, `/co purge t:30d #optimize`) will also optimize supported database tables and reclaim unused disk space. How this option is handled depends on the database backend:
 
 * SQLite already rebuilds the database from retained data and reclaims unused file space as part of a manual purge, so `#optimize` is not needed.
-* MySQL normally deletes matching rows. Adding `#optimize` also optimizes its tables to reclaim unused space.
+* MySQL normally deletes matching rows. Adding `#optimize` also removes saved entity data that no entity kill references anymore (this requires the `CREATE TEMPORARY TABLES` privilege), then optimizes its tables to reclaim unused space.
 * DuckDB deletes matching rows in one transaction and checkpoints afterward. `#optimize` has no additional effect.
 * ClickHouse drops fully covered monthly partitions for an unfiltered time purge and synchronously removes rows from partial or filtered partitions. Adding `#optimize` also runs `OPTIMIZE TABLE ... FINAL`.
 
diff --git a/lang/en.yml b/lang/en.yml
index e594dbf6d..d42120dc9 100644
--- a/lang/en.yml
+++ b/lang/en.yml
@@ -77,7 +77,9 @@ HELP_PURGE_1: "Delete data older than specified time."
 HELP_PURGE_2: "For example, \"{0}\" will delete all data older than one month, and only keep the last 30 days of data."
 HELP_PURGE_3: "Specify the amount of time to purge."
 HELP_PURGE_4: "Specify a world to limit the purge to."
-HELP_PURGE_5: "Include specific blocks in the purge."
+HELP_PURGE_5: "Include specific blocks or entity kills in the purge."
+HELP_PURGE_6: "Keep entity kills of these types."
+HELP_PURGE_7: "Only purge entity kills."
 HELP_PURGE_COMMAND: "Delete old block data."
 HELP_RADIUS_1: "Specify a radius area."
 HELP_RADIUS_2: "Examples: [r:10] (Only make changes within 10 blocks of you)"
diff --git a/src/main/java/net/coreprotect/command/HelpCommand.java b/src/main/java/net/coreprotect/command/HelpCommand.java
index 94b925bf8..72c35f7a2 100755
--- a/src/main/java/net/coreprotect/command/HelpCommand.java
+++ b/src/main/java/net/coreprotect/command/HelpCommand.java
@@ -72,6 +72,8 @@ else if (helpcommand.equals("purge") || helpcommand.equals("purges")) {
                     Chat.sendMessage(player, Color.DARK_AQUA + "| " + Color.GREY + "t:<time> " + Color.WHITE + "- " + Phrase.build(Phrase.HELP_PURGE_3));
                     Chat.sendMessage(player, Color.DARK_AQUA + "| " + Color.GREY + "r:<world> " + Color.WHITE + "- " + Phrase.build(Phrase.HELP_PURGE_4));
                     Chat.sendMessage(player, Color.DARK_AQUA + "| " + Color.GREY + "i:<include> " + Color.WHITE + "- " + Phrase.build(Phrase.HELP_PURGE_5));
+                    Chat.sendMessage(player, Color.DARK_AQUA + "| " + Color.GREY + "e:<exclude> " + Color.WHITE + "- " + Phrase.build(Phrase.HELP_PURGE_6));
+                    Chat.sendMessage(player, Color.DARK_AQUA + "| " + Color.GREY + "a:kill " + Color.WHITE + "- " + Phrase.build(Phrase.HELP_PURGE_7));
                     Chat.sendMessage(player, Color.GREY + Color.ITALIC + Phrase.build(Phrase.HELP_PURGE_2, "/co purge t:30d"));
                 }
                 else if (helpcommand.equals("reload")) {
diff --git a/src/main/java/net/coreprotect/command/PurgeCommand.java b/src/main/java/net/coreprotect/command/PurgeCommand.java
index 40a8cdcb3..b2401fd3f 100755
--- a/src/main/java/net/coreprotect/command/PurgeCommand.java
+++ b/src/main/java/net/coreprotect/command/PurgeCommand.java
@@ -13,6 +13,7 @@
 import java.text.NumberFormat;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
@@ -28,6 +29,7 @@
 import net.coreprotect.config.ConfigHandler;
 import net.coreprotect.consumer.Consumer;
 import net.coreprotect.database.Database;
+import net.coreprotect.database.PurgeFilter;
 import net.coreprotect.database.PurgePolicy;
 import net.coreprotect.language.Phrase;
 import net.coreprotect.language.Selector;
@@ -105,7 +107,7 @@ private static void reportPurgeFailure(Exception exception) throws Exception {
         ErrorReporter.report(exception);
     }
 
-    private static String findUnsupportedPurgeArgument(String[] args) {
+    static String findUnsupportedPurgeArgument(String[] args) {
         boolean includeContinuation = false;
         for (int i = 1; i < args.length; i++) {
             String token = args[i].trim();
@@ -132,6 +134,20 @@ private static String findUnsupportedPurgeArgument(String[] args) {
                 continue;
             }
 
+            if (argument.startsWith("e:") || argument.startsWith("exclude:")) {
+                String excludeValues = argument.replaceAll("exclude:", "").replaceAll("e:", "");
+                includeContinuation = excludeValues.length() == 0 || excludeValues.endsWith(",");
+                continue;
+            }
+
+            if (argument.startsWith("a:") || argument.startsWith("action:")) {
+                // Only kills can be purged by action; any other value must not fall through to an unrestricted purge.
+                if (!CommandParser.parseAction(new String[] { "purge", argument }).equals(Collections.singletonList(LookupActions.ENTITY_KILL))) {
+                    return token;
+                }
+                continue;
+            }
+
             if (argument.startsWith("t:") || argument.startsWith("time:")) {
                 continue;
             }
@@ -158,7 +174,7 @@ protected static void runCommand(final CommandSender player, boolean permission,
         final List<String> argExcludeUsers = CommandParser.parseExcludedUsers(player, args);
         final long[] argTime = CommandParser.parseTime(args);
         final int argWid = CommandParser.parseWorld(args, false, false);
-        final List<Integer> supportedActions = Arrays.asList();
+        final List<Integer> supportedActions = Arrays.asList(LookupActions.ENTITY_KILL);
         long startTime = argTime[1] > 0 ? argTime[0] : 0;
         long endTime = argTime[1] > 0 ? argTime[1] : argTime[0];
 
@@ -218,18 +234,14 @@ else if (endTime < 86400) {
         }
 
         StringBuilder restrict = new StringBuilder();
-        String includeBlock = "";
         List<Integer> includeBlockIds = new ArrayList<>();
-        String includeEntity = "";
+        List<Integer> includeEntityIds = new ArrayList<>();
         boolean hasBlock = false;
         boolean item = false;
         boolean entity = false;
         int restrictCount = 0;
 
         if (argBlocks.size() > 0) {
-            StringBuilder includeListMaterial = new StringBuilder();
-            StringBuilder includeListEntity = new StringBuilder();
-
             for (Object restrictTarget : argBlocks) {
                 String targetName = "";
 
@@ -237,17 +249,10 @@ else if (endTime < 86400) {
                     targetName = ((Material) restrictTarget).name();
                     int blockId = MaterialUtils.getBlockId(targetName, false);
                     includeBlockIds.add(blockId);
-                    if (includeListMaterial.length() == 0) {
-                        includeListMaterial = includeListMaterial.append(blockId);
-                    }
-                    else {
-                        includeListMaterial.append(",").append(blockId);
-                    }
 
                     /* Include legacy IDs */
                     int legacyId = BukkitAdapter.ADAPTER.getLegacyBlockId((Material) restrictTarget);
                     if (legacyId > 0) {
-                        includeListMaterial.append(",").append(legacyId);
                         includeBlockIds.add(legacyId);
                     }
 
@@ -256,26 +261,13 @@ else if (endTime < 86400) {
                     hasBlock = true;
                 }
                 else if (restrictTarget instanceof EntityType) {
-                    targetName = ((EntityType) restrictTarget).name();
-                    if (includeListEntity.length() == 0) {
-                        includeListEntity = includeListEntity.append(EntityUtils.getEntityId(targetName, false));
-                    }
-                    else {
-                        includeListEntity.append(",").append(EntityUtils.getEntityId(targetName, false));
-                    }
-
+                    includeEntityIds.add(EntityUtils.getEntityId(((EntityType) restrictTarget).name(), false));
                     targetName = ((EntityType) restrictTarget).name().toLowerCase(Locale.ROOT);
                     entity = true;
                 }
                 else if (restrictTarget instanceof String) {
                     int blockId = MaterialUtils.getBlockId((String) restrictTarget, false);
                     includeBlockIds.add(blockId);
-                    if (includeListMaterial.length() == 0) {
-                        includeListMaterial = includeListMaterial.append(blockId);
-                    }
-                    else {
-                        includeListMaterial.append(",").append(blockId);
-                    }
 
                     targetName = ((String) restrictTarget).toLowerCase(Locale.ROOT);
                     hasBlock = true;
@@ -290,12 +282,28 @@ else if (restrictTarget instanceof String) {
 
                 restrictCount++;
             }
+        }
 
-            includeBlock = includeListMaterial.toString();
-            includeEntity = includeListEntity.toString();
+        List<Integer> excludeEntityIds = new ArrayList<>();
+        StringBuilder exclude = new StringBuilder();
+        for (Object excludeTarget : argExclude.keySet()) {
+            if (!(excludeTarget instanceof EntityType)) {
+                // Purge exclusions only support entity types (kills to keep).
+                Chat.sendMessage(player, Color.DARK_AQUA + "CoreProtect " + Color.WHITE + "- " + Phrase.build(Phrase.INVALID_PARAMETER, "e:" + excludeTarget.toString().toLowerCase(Locale.ROOT)));
+                return;
+            }
+            excludeEntityIds.add(EntityUtils.getEntityId(((EntityType) excludeTarget).name(), false));
+            exclude.append(exclude.length() == 0 ? "" : ", ").append(((EntityType) excludeTarget).name().toLowerCase(Locale.ROOT));
+        }
+        if (!argExcludeUsers.isEmpty()) {
+            Chat.sendMessage(player, Color.DARK_AQUA + "CoreProtect " + Color.WHITE + "- " + Phrase.build(Phrase.INVALID_PARAMETER, "e:" + argExcludeUsers.get(0)));
+            return;
         }
 
-        if (entity) {
+        boolean killsOnly = argAction.contains(LookupActions.ENTITY_KILL);
+        boolean entityFilter = entity || killsOnly || !excludeEntityIds.isEmpty();
+        boolean excludeWithoutKills = !excludeEntityIds.isEmpty() && hasBlock && !entity; // a block restriction keeps every kill
+        if ((killsOnly && hasBlock) || (entity && !excludeEntityIds.isEmpty()) || excludeWithoutKills || (entityFilter && ConfigHandler.databaseType.isClickHouse())) {
             Chat.sendMessage(player, Color.DARK_AQUA + "CoreProtect " + Color.WHITE + "- " + Phrase.build(Phrase.ACTION_NOT_SUPPORTED));
             return;
         }
@@ -309,11 +317,21 @@ else if (restrictTarget instanceof String) {
         }
 
         final StringBuilder restrictTargets = restrict;
-        final String includeBlockFinal = includeBlock;
+        final StringBuilder excludeTargets = exclude;
         final List<Integer> includeBlockIdsFinal = List.copyOf(includeBlockIds);
+        final List<Integer> includeEntityIdsFinal = List.copyOf(includeEntityIds);
+        final List<Integer> excludeEntityIdsFinal = List.copyOf(excludeEntityIds);
+        final boolean killsOnlyFinal = killsOnly;
         final boolean optimize = optimizeCheck;
-        final boolean hasBlockRestriction = hasBlock;
         final int restrictCountFinal = restrictCount;
+        String restrictSelector = Selector.FIRST; // block
+        if (hasBlock && entity) {
+            restrictSelector = Selector.THIRD; // target
+        }
+        else if (entity) {
+            restrictSelector = Selector.SECOND; // entity
+        }
+        final String restrictKind = restrictSelector;
 
         class BasicThread implements Runnable {
 
@@ -340,6 +358,7 @@ public void run() {
                     long timeStart = startTime > 0 ? (timestamp - startTime) : 0;
                     long timeEnd = timestamp - endTime;
                     long removed = 0;
+                    PurgeFilter purgeFilter = new PurgeFilter(timeStart, timeEnd, argWid, includeBlockIdsFinal, includeEntityIdsFinal, excludeEntityIdsFinal, killsOnlyFinal);
 
                     for (int i = 0; i <= 5; i++) {
                         requirePurgeNotCancelled();
@@ -382,8 +401,11 @@ public void run() {
                         Chat.sendGlobalMessage(player, Phrase.build(Phrase.PURGE_STARTED, "#global"));
                     }
 
-                    if (hasBlockRestriction) {
-                        Chat.sendGlobalMessage(player, Phrase.build(Phrase.ROLLBACK_INCLUDE, restrictTargets.toString(), Selector.FIRST, Selector.FIRST, (restrictCountFinal == 1 ? Selector.FIRST : Selector.SECOND))); // include
+                    if (restrictCountFinal > 0) {
+                        Chat.sendGlobalMessage(player, Phrase.build(Phrase.ROLLBACK_INCLUDE, restrictTargets.toString(), Selector.FIRST, restrictKind, (restrictCountFinal == 1 ? Selector.FIRST : Selector.SECOND))); // include
+                    }
+                    if (excludeEntityIdsFinal.size() > 0) {
+                        Chat.sendGlobalMessage(player, Phrase.build(Phrase.ROLLBACK_INCLUDE, excludeTargets.toString(), Selector.SECOND, Selector.SECOND, (excludeEntityIdsFinal.size() == 1 ? Selector.FIRST : Selector.SECOND))); // exclude
                     }
 
                     Chat.sendGlobalMessage(player, Phrase.build(Phrase.PURGE_NOTICE_1));
@@ -492,33 +514,17 @@ public void run() {
                             boolean error = false;
                             if (!excludeTables.contains(table)) {
                                 try {
-                                    boolean purge = true;
                                     String timeLimit = "";
                                     if (table.equals("entity_spawn")) {
                                         timeLimit = " WHERE removed=0 OR block_rowid IN(SELECT rowid FROM " + purgePrefix + "block) OR kill_rowid IN(SELECT rowid FROM " + purgePrefix + "entity) OR rowid IN(SELECT entity_spawn_rowid FROM " + purgePrefix + "entity_container) OR rowid IN(SELECT entity_spawn_rowid FROM " + purgePrefix + "entity_interaction)";
                                     }
-                                    else if (PurgePolicy.isPurgeable(table)) {
-                                        String blockRestriction = "(";
-                                        if (hasBlockRestriction && PurgePolicy.supportsBlockRestriction(table)) {
-                                            blockRestriction = "action IN(" + LookupActions.ENTITY_KILL + "," + LookupActions.ENTITY_SPAWN + ") OR type NOT IN(" + includeBlockFinal + ") OR (type IN(" + includeBlockFinal + ") AND ";
-                                        }
-                                        else if (hasBlockRestriction) {
-                                            purge = false;
-                                        }
-
-                                        if (argWid > 0 && PurgePolicy.isWorldScoped(table)) {
-                                            if (table.equals("entity_container") || table.equals("entity_interaction")) {
-                                                if (purge) {
-                                                    String worldMatch = "(wid = '" + argWid + "' OR entity_spawn_rowid IN(SELECT rowid FROM " + ConfigHandler.prefix + "entity_spawn WHERE current_wid = '" + argWid + "'))";
-                                                    timeLimit = " WHERE (" + worldMatch + " AND (time >= '" + timeEnd + "' OR time < '" + timeStart + "')) OR NOT " + worldMatch;
-                                                }
-                                            }
-                                            else {
-                                                timeLimit = " WHERE (" + blockRestriction + "wid = '" + argWid + "' AND (time >= '" + timeEnd + "' OR time < '" + timeStart + "'))) OR (wid != '" + argWid + "')";
-                                            }
-                                        }
-                                        else if (argWid == 0 && purge) {
-                                            timeLimit = " WHERE " + blockRestriction + "(time >= '" + timeEnd + "' OR time < '" + timeStart + "'))";
+                                    else if (table.equals("entity")) {
+                                        timeLimit = " WHERE " + PurgeFilter.entityRetainCondition(purgePrefix + "block");
+                                    }
+                                    else {
+                                        String purgeCondition = purgeFilter.deleteCondition(table, ConfigHandler.prefix);
+                                        if (purgeCondition != null) {
+                                            timeLimit = " WHERE NOT (" + purgeCondition + ")";
                                         }
                                     }
                                     query = "INSERT INTO " + purgePrefix + table + insertColumns + " SELECT " + selectColumns + " FROM " + ConfigHandler.prefix + table + timeLimit;
@@ -571,35 +577,21 @@ else if (argWid == 0 && purge) {
                                 }
 
                                 try {
-                                    boolean purge = PurgePolicy.isPurgeable(table);
-
-                                    String blockRestriction = "";
-                                    if (hasBlockRestriction && PurgePolicy.supportsBlockRestriction(table)) {
-                                        blockRestriction = "action NOT IN(" + LookupActions.ENTITY_KILL + "," + LookupActions.ENTITY_SPAWN + ") AND type IN(" + includeBlockFinal + ") AND ";
-                                    }
-                                    else if (hasBlockRestriction) {
-                                        purge = false;
-                                    }
-
-                                    String worldRestriction = "";
-                                    if (argWid > 0 && PurgePolicy.isWorldScoped(table)) {
-                                        if (table.equals("entity_container") || table.equals("entity_interaction")) {
-                                            worldRestriction = " AND (wid = '" + argWid + "' OR entity_spawn_rowid IN(SELECT rowid FROM " + ConfigHandler.prefix + "entity_spawn WHERE current_wid = '" + argWid + "'))";
-                                        }
-                                        else {
-                                            worldRestriction = " AND wid = '" + argWid + "'";
-                                        }
-                                    }
-                                    else if (argWid > 0) {
-                                        purge = false;
-                                    }
-
-                                    if (purge) {
-                                        query = "DELETE FROM " + purgePrefix + table + " WHERE " + blockRestriction + "time < '" + timeEnd + "' AND time >= '" + timeStart + "'" + worldRestriction;
+                                    if (table.equals("entity")) {
+                                        query = PurgeFilter.deleteUnreferencedEntities(purgePrefix + "entity", purgePrefix + "block");
                                         preparedStmt = preparePurgeStatement(connection, query);
                                         preparedStmt.execute();
                                         preparedStmt.close();
                                     }
+                                    else {
+                                        String purgeCondition = purgeFilter.deleteCondition(table, ConfigHandler.prefix);
+                                        if (purgeCondition != null) {
+                                            query = "DELETE FROM " + purgePrefix + table + " WHERE " + purgeCondition;
+                                            preparedStmt = preparePurgeStatement(connection, query);
+                                            preparedStmt.execute();
+                                            preparedStmt.close();
+                                        }
+                                    }
                                 }
                                 catch (Exception e) {
                                     reportPurgeFailure(e);
@@ -643,31 +635,16 @@ else if (argWid > 0) {
 
                         if (!ConfigHandler.databaseType.isSQLite()) {
                             try {
-                                boolean purge = PurgePolicy.isPurgeable(table);
-
-                                String blockRestriction = "";
-                                if (hasBlockRestriction && PurgePolicy.supportsBlockRestriction(table)) {
-                                    blockRestriction = "action NOT IN(" + LookupActions.ENTITY_KILL + "," + LookupActions.ENTITY_SPAWN + ") AND type IN(" + includeBlockFinal + ") AND ";
-                                }
-                                else if (hasBlockRestriction) {
-                                    purge = false;
-                                }
-
-                                String worldRestriction = "";
-                                if (argWid > 0 && PurgePolicy.isWorldScoped(table)) {
-                                    if (table.equals("entity_container") || table.equals("entity_interaction")) {
-                                        worldRestriction = " AND (wid = '" + argWid + "' OR entity_spawn_rowid IN(SELECT rowid FROM " + ConfigHandler.prefix + "entity_spawn WHERE current_wid = '" + argWid + "'))";
-                                    }
-                                    else {
-                                        worldRestriction = " AND wid = '" + argWid + "'";
-                                    }
-                                }
-                                else if (argWid > 0) {
-                                    purge = false;
+                                String purgeCondition = purgeFilter.deleteCondition(table, ConfigHandler.prefix);
+                                if (purgeCondition != null && table.equals("block") && purgeFilter.removesKills() && !purgeFilter.purgesEntitiesByTime()) {
+                                    query = purgeFilter.deleteEntitiesOfPurgedKills(ConfigHandler.databaseType, ConfigHandler.prefix);
+                                    preparedStmt = preparePurgeStatement(connection, query);
+                                    removed = removed + preparedStmt.executeUpdate();
+                                    preparedStmt.close();
                                 }
 
-                                if (purge) {
-                                    query = "DELETE FROM " + ConfigHandler.prefix + table + " WHERE " + blockRestriction + "time < '" + timeEnd + "' AND time >= '" + timeStart + "'" + worldRestriction;
+                                if (purgeCondition != null) {
+                                    query = "DELETE FROM " + ConfigHandler.prefix + table + " WHERE " + purgeCondition;
                                     preparedStmt = preparePurgeStatement(connection, query);
                                     removed = removed + preparedStmt.executeUpdate();
                                     preparedStmt.close();
@@ -687,6 +664,32 @@ else if (argWid > 0) {
                         }
                     }
 
+                    if (ConfigHandler.databaseType.isMySQL() && optimize) {
+                        try {
+                            for (String sweepQuery : PurgeFilter.mysqlOrphanSweepSetup(ConfigHandler.prefix)) {
+                                preparedStmt = preparePurgeStatement(connection, sweepQuery);
+                                preparedStmt.execute();
+                                preparedStmt.close();
+                            }
+                            preparedStmt = preparePurgeStatement(connection, PurgeFilter.mysqlOrphanSweepDelete(ConfigHandler.prefix));
+                            removed = removed + preparedStmt.executeUpdate();
+                            preparedStmt.close();
+                        }
+                        catch (Exception e) {
+                            reportPurgeFailure(e);
+                        }
+                        finally {
+                            try {
+                                preparedStmt = preparePurgeStatement(connection, PurgeFilter.mysqlOrphanSweepTeardown(ConfigHandler.prefix));
+                                preparedStmt.execute();
+                                preparedStmt.close();
+                            }
+                            catch (Exception e) {
+                                reportPurgeFailure(e);
+                            }
+                        }
+                    }
+
                     requirePurgeNotCancelled();
                     String retainedPrefix = ConfigHandler.databaseType.isSQLite() ? purgePrefix : ConfigHandler.prefix;
                     query = "UPDATE " + retainedPrefix + "entity_spawn SET kill_rowid=NULL WHERE kill_rowid IS NOT NULL AND NOT EXISTS (SELECT 1 FROM " + retainedPrefix + "entity WHERE " + retainedPrefix + "entity.rowid=" + retainedPrefix + "entity_spawn.kill_rowid)";
diff --git a/src/main/java/net/coreprotect/command/TabHandler.java b/src/main/java/net/coreprotect/command/TabHandler.java
index 35501ca26..db5bd58de 100755
--- a/src/main/java/net/coreprotect/command/TabHandler.java
+++ b/src/main/java/net/coreprotect/command/TabHandler.java
@@ -447,7 +447,7 @@ private List<String> handleSecondArgCompletions(CommandSender sender, String cmd
             return StringUtil.copyPartialMatches(argument1, completions, new ArrayList<>(completions.size()));
         }
         else if (argument0.equals("purge") && sender.hasPermission("coreprotect.purge")) {
-            List<String> completions = new ArrayList<>(Arrays.asList("t:", "r:", "i:"));
+            List<String> completions = new ArrayList<>(Arrays.asList("t:", "r:", "i:", "e:", "a:kill"));
             return StringUtil.copyPartialMatches(argument1, completions, new ArrayList<>(completions.size()));
         }
         else if (hasLookupCommand(argument0, sender)) {
@@ -464,10 +464,10 @@ private List<String> handlePurgeThirdArgCompletions(String arg1, String arg2) {
         String argument2 = arg2.toLowerCase(Locale.ROOT);
 
         if (argument1.startsWith("t:")) {
-            List<String> completions = new ArrayList<>(Arrays.asList("r:", "i:"));
+            List<String> completions = new ArrayList<>(Arrays.asList("r:", "i:", "e:", "a:kill"));
             return StringUtil.copyPartialMatches(argument2, completions, new ArrayList<>(completions.size()));
         }
-        else if (argument1.startsWith("r:") || argument1.startsWith("i:")) {
+        else if (argument1.startsWith("r:") || argument1.startsWith("i:") || argument1.startsWith("e:") || argument1.startsWith("a:")) {
             List<String> completions = new ArrayList<>(Arrays.asList("t:"));
             return StringUtil.copyPartialMatches(argument2, completions, new ArrayList<>(completions.size()));
         }
diff --git a/src/main/java/net/coreprotect/database/PurgeFilter.java b/src/main/java/net/coreprotect/database/PurgeFilter.java
new file mode 100644
index 000000000..d98da93a8
--- /dev/null
+++ b/src/main/java/net/coreprotect/database/PurgeFilter.java
@@ -0,0 +1,241 @@
+package net.coreprotect.database;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.StringJoiner;
+
+import net.coreprotect.model.action.LookupActions;
+
+/**
+ * Selects the rows that one /co purge removes on the relational backends (SQLite, MySQL, DuckDB).
+ *
+ * <p>
+ * co_entity has no world or type column: each row belongs to the co_block kill row whose data column holds its
+ * rowid. The statements built here remove co_entity rows through those kill rows, so a scoped purge never leaves
+ * entity data behind without the kill row that references it. Player kills also use the kill action, but they
+ * store type 0 and a user id in data, so they are never treated as co_entity references.
+ */
+public final class PurgeFilter {
+
+    private final long timeStart;
+    private final long timeEnd;
+    private final int worldId;
+    private final List<Integer> blockTypes;
+    private final List<Integer> entityTypes;
+    private final List<Integer> excludedEntityTypes;
+    private final boolean killsOnly;
+
+    /**
+     * @param timeStart
+     *            the oldest time (inclusive) to purge
+     * @param timeEnd
+     *            the newest time (exclusive) to purge
+     * @param worldId
+     *            the world to purge, or 0 for every world
+     * @param blockTypes
+     *            material ids (i:) that restrict the purge to those co_block rows, or an empty list
+     * @param entityTypes
+     *            entity type ids (i:) that restrict the purge to kills of those types, or an empty list
+     * @param excludedEntityTypes
+     *            entity type ids (e:) whose kills the purge keeps, or an empty list
+     * @param killsOnly
+     *            true (a:kill) to restrict the purge to kill rows
+     */
+    public PurgeFilter(long timeStart, long timeEnd, int worldId, List<Integer> blockTypes, List<Integer> entityTypes, List<Integer> excludedEntityTypes, boolean killsOnly) {
+        this.timeStart = timeStart;
+        this.timeEnd = timeEnd;
+        this.worldId = worldId;
+        this.blockTypes = List.copyOf(blockTypes);
+        this.entityTypes = List.copyOf(entityTypes);
+        this.excludedEntityTypes = List.copyOf(excludedEntityTypes);
+        this.killsOnly = killsOnly;
+    }
+
+    /**
+     * Returns whether the purge is limited to selected co_block rows, which leaves the other tables untouched.
+     */
+    private boolean restrictsTables() {
+        return !blockTypes.isEmpty() || !entityTypes.isEmpty() || killsOnly;
+    }
+
+    /**
+     * Returns whether the purge removes co_block kill rows. A restriction to materials alone keeps every kill row.
+     */
+    public boolean removesKills() {
+        return blockTypes.isEmpty() || !entityTypes.isEmpty() || killsOnly;
+    }
+
+    /**
+     * Returns whether co_entity rows can be removed by time alone. This holds when every kill row in the time range
+     * is purged, because a co_entity row always has the same time as its kill row.
+     */
+    public boolean purgesEntitiesByTime() {
+        return worldId <= 0 && !restrictsTables() && excludedEntityTypes.isEmpty();
+    }
+
+    /**
+     * Builds the condition that matches the rows this purge removes from a table.
+     *
+     * @param table
+     *            the table name without prefix
+     * @param prefix
+     *            the prefix of the tables that subqueries read from
+     * @return the SQL condition, or null when the purge keeps every row of the table
+     */
+    public String deleteCondition(String table, String prefix) {
+        if (table.equals("block")) {
+            return blockCondition("");
+        }
+        if (table.equals("entity")) {
+            return purgesEntitiesByTime() ? timeCondition("") : null;
+        }
+        if (!PurgePolicy.isPurgeable(table) || restrictsTables()) {
+            return null;
+        }
+        if (worldId <= 0) {
+            return timeCondition("");
+        }
+        if (!PurgePolicy.isWorldScoped(table)) {
+            return null;
+        }
+        if (table.equals("entity_container") || table.equals("entity_interaction")) {
+            return timeCondition("") + " AND (wid = " + worldId + " OR entity_spawn_rowid IN(SELECT rowid FROM " + prefix + "entity_spawn WHERE current_wid = " + worldId + "))";
+        }
+        return timeCondition("") + " AND wid = " + worldId;
+    }
+
+    /**
+     * Builds the condition that matches the co_block rows this purge removes.
+     *
+     * @param qualifier
+     *            the table alias to prefix each column with, such as "b.", or an empty string
+     * @return the SQL condition
+     */
+    private String blockCondition(String qualifier) {
+        StringBuilder condition = new StringBuilder(timeCondition(qualifier));
+        if (worldId > 0) {
+            condition.append(" AND ").append(qualifier).append("wid = ").append(worldId);
+        }
+
+        List<String> restrictions = new ArrayList<>();
+        if (!blockTypes.isEmpty()) {
+            restrictions.add("(" + qualifier + "action NOT IN(" + LookupActions.ENTITY_KILL + "," + LookupActions.ENTITY_SPAWN + ") AND " + qualifier + "type IN(" + joinIds(blockTypes) + "))");
+        }
+        if (!entityTypes.isEmpty()) {
+            restrictions.add("(" + qualifier + "action = " + LookupActions.ENTITY_KILL + " AND " + qualifier + "type IN(" + joinIds(entityTypes) + "))");
+        }
+        else if (killsOnly) {
+            restrictions.add(qualifier + "action = " + LookupActions.ENTITY_KILL);
+        }
+        if (!restrictions.isEmpty()) {
+            condition.append(" AND (").append(String.join(" OR ", restrictions)).append(")");
+        }
+
+        if (!excludedEntityTypes.isEmpty()) {
+            condition.append(" AND NOT (").append(qualifier).append("action = ").append(LookupActions.ENTITY_KILL).append(" AND ").append(qualifier).append("type IN(").append(joinIds(excludedEntityTypes)).append("))");
+        }
+        return condition.toString();
+    }
+
+    /**
+     * Builds the MySQL or DuckDB statement that deletes the co_entity rows of the kill rows this purge removes. Run it
+     * before the co_block delete, while those kill rows still exist.
+     *
+     * @param databaseType
+     *            the MySQL or DuckDB backend
+     * @param prefix
+     *            the table prefix
+     * @return the SQL statement
+     */
+    public String deleteEntitiesOfPurgedKills(DatabaseType databaseType, String prefix) {
+        if (databaseType.isMySQL()) {
+            // The join form avoids a dependent subquery on MySQL 5.7 and MariaDB.
+            return "DELETE e FROM " + prefix + "entity AS e INNER JOIN " + prefix + "block AS b ON b.data = e.rowid WHERE " + killReference("b.") + " AND " + blockCondition("b.");
+        }
+        return "DELETE FROM " + prefix + "entity WHERE rowid IN(SELECT data FROM " + prefix + "block WHERE " + killReference("") + " AND " + blockCondition("") + ")";
+    }
+
+    /**
+     * Builds the SQLite copy condition that keeps the co_entity rows still referenced by a retained kill row.
+     *
+     * @param retainedBlockTable
+     *            the co_block table that holds the rows the purge keeps
+     * @return the SQL condition
+     */
+    public static String entityRetainCondition(String retainedBlockTable) {
+        return "rowid IN(" + killReferences(retainedBlockTable) + ")";
+    }
+
+    /**
+     * Builds the statement that deletes co_entity rows that no kill row references.
+     *
+     * @param entityTable
+     *            the co_entity table to clean
+     * @param blockTable
+     *            the co_block table that holds the kill rows
+     * @return the SQL statement
+     */
+    public static String deleteUnreferencedEntities(String entityTable, String blockTable) {
+        return "DELETE FROM " + entityTable + " WHERE rowid NOT IN(" + killReferences(blockTable) + ")";
+    }
+
+    /**
+     * Builds the MySQL statements that prepare an orphan sweep: a temporary table of the co_entity ids that kill rows
+     * reference. The table avoids an anti-join on the unindexed co_block.data column.
+     *
+     * @param prefix
+     *            the table prefix
+     * @return the SQL statements, in order
+     */
+    public static List<String> mysqlOrphanSweepSetup(String prefix) {
+        String keepTable = prefix + "entity_keep";
+        List<String> statements = new ArrayList<>();
+        statements.add("DROP TEMPORARY TABLE IF EXISTS " + keepTable);
+        statements.add("CREATE TEMPORARY TABLE " + keepTable + " (rowid INT NOT NULL PRIMARY KEY) ENGINE=InnoDB");
+        statements.add("INSERT IGNORE INTO " + keepTable + " (rowid) " + killReferences(prefix + "block"));
+        return statements;
+    }
+
+    /**
+     * Builds the MySQL statement that deletes co_entity rows no kill row references, such as rows left by earlier
+     * world purges. Run {@link #mysqlOrphanSweepSetup(String)} first.
+     *
+     * @param prefix
+     *            the table prefix
+     * @return the SQL statement
+     */
+    public static String mysqlOrphanSweepDelete(String prefix) {
+        return "DELETE e FROM " + prefix + "entity AS e LEFT JOIN " + prefix + "entity_keep AS k ON k.rowid = e.rowid WHERE k.rowid IS NULL";
+    }
+
+    /**
+     * Builds the MySQL statement that drops the temporary table of an orphan sweep.
+     *
+     * @param prefix
+     *            the table prefix
+     * @return the SQL statement
+     */
+    public static String mysqlOrphanSweepTeardown(String prefix) {
+        return "DROP TEMPORARY TABLE IF EXISTS " + prefix + "entity_keep";
+    }
+
+    private String timeCondition(String qualifier) {
+        return qualifier + "time < " + timeEnd + " AND " + qualifier + "time >= " + timeStart;
+    }
+
+    private static String killReferences(String blockTable) {
+        return "SELECT data FROM " + blockTable + " WHERE " + killReference("") + " AND data IS NOT NULL";
+    }
+
+    private static String killReference(String qualifier) {
+        return qualifier + "action = " + LookupActions.ENTITY_KILL + " AND " + qualifier + "type <> 0";
+    }
+
+    private static String joinIds(List<Integer> ids) {
+        StringJoiner joiner = new StringJoiner(",");
+        for (Integer id : ids) {
+            joiner.add(String.valueOf(id));
+        }
+        return joiner.toString();
+    }
+}
diff --git a/src/main/java/net/coreprotect/language/Language.java b/src/main/java/net/coreprotect/language/Language.java
index 403b23c84..4323c3cc1 100644
--- a/src/main/java/net/coreprotect/language/Language.java
+++ b/src/main/java/net/coreprotect/language/Language.java
@@ -114,7 +114,9 @@ public static void loadPhrases() {
         phrases.put(Phrase.HELP_PURGE_2, "For example, \"{0}\" will delete all data older than one month, and only keep the last 30 days of data.");
         phrases.put(Phrase.HELP_PURGE_3, "Specify the amount of time to purge.");
         phrases.put(Phrase.HELP_PURGE_4, "Specify a world to limit the purge to.");
-        phrases.put(Phrase.HELP_PURGE_5, "Include specific blocks in the purge.");
+        phrases.put(Phrase.HELP_PURGE_5, "Include specific blocks or entity kills in the purge.");
+        phrases.put(Phrase.HELP_PURGE_6, "Keep entity kills of these types.");
+        phrases.put(Phrase.HELP_PURGE_7, "Only purge entity kills.");
         phrases.put(Phrase.HELP_PURGE_COMMAND, "Delete old block data.");
         phrases.put(Phrase.HELP_RADIUS_1, "Specify a radius area.");
         phrases.put(Phrase.HELP_RADIUS_2, "Examples: [r:10] (Only make changes within 10 blocks of you)");
diff --git a/src/main/java/net/coreprotect/language/Phrase.java b/src/main/java/net/coreprotect/language/Phrase.java
index 079cb09d4..6bbe56e31 100644
--- a/src/main/java/net/coreprotect/language/Phrase.java
+++ b/src/main/java/net/coreprotect/language/Phrase.java
@@ -90,6 +90,8 @@ public enum Phrase {
     HELP_PURGE_3,
     HELP_PURGE_4,
     HELP_PURGE_5,
+    HELP_PURGE_6,
+    HELP_PURGE_7,
     HELP_PURGE_COMMAND,
     HELP_RADIUS_1,
     HELP_RADIUS_2,