Improved multi-server support for columnar storage

docs/auto-purge.md

@@ -33,7 +33,9 @@ Automatic purging runs in the background. SQLite, MySQL, and DuckDB remove old r
 
 Automatic purging does not rebuild SQLite, optimize MySQL, or run a ClickHouse `OPTIMIZE FINAL`. DuckDB performs a checkpoint after cleanup. Deleting rows helps control future growth, but it may not immediately reduce the database file or table size on disk.
 
-Only one automatic purge can run at a time. If the server shuts down, a manual purge starts, a database migration or conversion starts, or the consumer is manually paused, the automatic purge stops safely and can continue during the next scheduled run.
+Only one automatic purge can run per CoreProtect instance at a time. If the server shuts down, a manual purge starts, a database migration or conversion starts, or the consumer is manually paused, the automatic purge stops safely and can continue during the next scheduled run.
+
+ClickHouse rejects automatic and manual purges while `database-lock` is disabled. For a shared namespace, stop every installation, enable `database-lock` on one installation, restart or reload it, and finish the purge before restoring shared-writer mode.
 
 ## Important Notes
 

docs/config.md

@@ -49,7 +49,7 @@ clickhouse-tls: false
 
 Create the configured database first with a database engine that provides persistent UUID-backed table identities. `Atomic` is the normal choice for self-hosted ClickHouse and is the default there; for example, `CREATE DATABASE coreprotect ENGINE = Atomic`. CoreProtect checks this capability before creating any tables, then creates and validates its prefixed tables and views. The account needs permission to read `system.databases` and `system.tables`, create and read/write the CoreProtect objects, run mutations and partition drops, and drop or truncate purge-target tables. Server-wide database-creation permission is not required; `#optimize` additionally requires table optimization permission.
 
-Only one active CoreProtect installation may use each ClickHouse database and prefix. Keep `database-lock` enabled and preserve `plugins/CoreProtect/.clickhouse-writer` with the installation; never copy it to another active server. Replicated or distributed tables and multiple active writers are not supported.
+Multiple active CoreProtect installations may share one ClickHouse database and prefix when `database-lock` is disabled. Configure every installation to use the same CoreProtect version, direct endpoint to the same physical ClickHouse server, and prefix, with a separate CoreProtect data directory; do not use a load balancer across independent nodes. Keep their system clocks synchronized; events recorded by different installations in the same second have an unspecified relative order. A shared prefix is one logical namespace: use globally unique world names unless same-named worlds are intentionally the same world, because their coordinate history is combined. Player attribution assumes that a username is not reassigned to a different account within that namespace and that a UUID-bearing login precedes UUID-less activity under a changed name. UUID-bearing observations preserve an account ID through ordinary username changes, but simultaneous username changes and username reuse are outside the base shared-writer contract. A server may continue displaying a peer's previously cached username until it observes that player locally or CoreProtect reloads. ClickHouse has no cross-server maintenance fence without Keeper, so every purge is rejected while `database-lock` is disabled. To purge a shared namespace, stop every installation sharing it, enable `database-lock` on the one installation that will run the purge, and restart or reload that installation before purging. Follow the same stop-all rule before a database migration; the migration command updates only the installation that runs it, so update the other configurations before restarting them. Replicated and distributed tables remain unsupported.
 
 ## Per-World Configuration
 

docs/database-migration.md

@@ -68,7 +68,7 @@ Source history rows are not deleted or replaced. Migration may update operationa
 5. Resume normal activity and test a newly logged change.
 6. Keep the source backup until the migrated database has been verified. Archive or remove it only when you are satisfied that the migration is complete.
 
-When either the source or target is ClickHouse, preserve `plugins/CoreProtect/.clickhouse-writer` with the installation so the retained ClickHouse dataset remains recoverable. Do not copy it to another active server or replace it after migration.
+When a ClickHouse source or target namespace is shared, stop every peer first and run the migration from exactly one installation with `database-lock` enabled. Before restarting any peer that will use the migrated target namespace, update its database selector, connection, and prefix to match the migrated installation.
 
 ## Failure and Recovery
 
@@ -90,7 +90,7 @@ If target activation or the atomic `config.yml` update fails after verification,
 * Confirm that the target settings were loaded while `database-type` still selected the source.
 * Verify the target host, port, database, credentials, table prefix, and available storage.
 * Confirm that the target account can create, read, insert, update or mutate, and delete or drop the required objects.
-* For ClickHouse, confirm the server is version 25.6 or newer and that this installation owns the matching `.clickhouse-writer` identity.
+* For ClickHouse, confirm the server is version 25.6 or newer, no other process is using the same CoreProtect data directory, and every other CoreProtect installation sharing the source or target namespace is stopped.
 
 ### Verification Failure or Interruption
 

src/main/java/net/coreprotect/api/BlockAPI.java

@@ -75,7 +75,7 @@ public class BlockAPI {
             try (Statement statement = connection.createStatement()) {
                 String table = DuckDBLookupQuery.spatialTable(connection, "block", worldId, x, x, z, z, "spatial_rows");
                 String index = ConfigHandler.databaseType.isDuckDB() ? "" : WorldUtils.getWidIndex("block");
-                String query = "SELECT time," + ConfigHandler.databaseType.getUserColumn() + ",action,type,data,blockdata,rolled_back FROM " + table + " " + index + "WHERE wid = " + worldId + " AND x = " + x + " AND z = " + z + " AND y = " + y + " AND time > " + checkTime + " ORDER BY rowid DESC";
+                String query = "SELECT time," + ConfigHandler.databaseType.getUserColumn() + ",action,type,data,blockdata,rolled_back FROM " + table + " " + index + "WHERE wid = " + worldId + " AND x = " + x + " AND z = " + z + " AND y = " + y + " AND time > " + checkTime + " ORDER BY " + ConfigHandler.getDescendingEventOrder();
 
                 try (ResultSet results = statement.executeQuery(query)) {
                     while (results.next()) {
@@ -158,7 +158,7 @@ public class BlockAPI {
             if (userId != null) {
                 query.append(" AND ").append(ConfigHandler.databaseType.getUserColumn()).append(" = ?");
             }
-            query.append(" ORDER BY rowid DESC");
+            query.append(" ORDER BY ").append(ConfigHandler.getDescendingEventOrder());
             if (options.hasLimit()) {
                 query.append(" LIMIT ").append(options.getLimitCount()).append(" OFFSET ").append(options.getLimitOffset());
             }

src/main/java/net/coreprotect/api/ItemAPI.java

@@ -54,7 +54,7 @@ public class ItemAPI {
                     .append(ItemLogger.ITEM_CREATE).append(",")
                     .append(ItemLogger.ITEM_SELL).append(",")
                     .append(ItemLogger.ITEM_BUY).append(")");
-            query.append(" ORDER BY rowid DESC");
+            query.append(" ORDER BY ").append(ConfigHandler.getDescendingEventOrder());
             filter.appendLimit(query);
 
             try (PreparedStatement statement = connection.prepareStatement(query.toString())) {

src/main/java/net/coreprotect/api/MessageAPI.java

@@ -61,7 +61,7 @@ public class MessageAPI {
                 query.append(WorldUtils.getWidIndex(table));
             }
             filter.appendWhere(query);
-            query.append(" ORDER BY rowid DESC");
+            query.append(" ORDER BY ").append(ConfigHandler.getDescendingEventOrder());
             filter.appendLimit(query);
 
             try (PreparedStatement statement = connection.prepareStatement(query.toString())) {

src/main/java/net/coreprotect/api/SessionLookup.java

@@ -119,7 +119,7 @@ public class SessionLookup {
                 query.append(WorldUtils.getWidIndex("session"));
             }
             filter.appendWhere(query);
-            query.append(" ORDER BY rowid DESC");
+            query.append(" ORDER BY ").append(ConfigHandler.getDescendingEventOrder());
             filter.appendLimit(query);
 
             try (PreparedStatement statement = connection.prepareStatement(query.toString())) {
@@ -162,7 +162,7 @@ public class SessionLookup {
      * @return The SQL query string
      */
     private static String buildSessionQuery(int userId, int checkTime) {
-        return "SELECT time," + ConfigHandler.databaseType.getUserColumn() + ",wid,x,y,z,action FROM " + ConfigHandler.prefix + "session WHERE " + ConfigHandler.databaseType.getUserColumn() + " = " + userId + " AND time > " + checkTime + " ORDER BY rowid DESC";
+        return "SELECT time," + ConfigHandler.databaseType.getUserColumn() + ",wid,x,y,z,action FROM " + ConfigHandler.prefix + "session WHERE " + ConfigHandler.databaseType.getUserColumn() + " = " + userId + " AND time > " + checkTime + " ORDER BY " + ConfigHandler.getDescendingEventOrder();
     }
 
     /**

src/main/java/net/coreprotect/api/SignAPI.java

@@ -58,7 +58,7 @@ public class SignAPI {
             }
             filter.appendWhere(query);
             query.append(" AND action = '").append(SignActions.PLACE).append("' AND (LENGTH(line_1) > 0 OR LENGTH(line_2) > 0 OR LENGTH(line_3) > 0 OR LENGTH(line_4) > 0 OR LENGTH(line_5) > 0 OR LENGTH(line_6) > 0 OR LENGTH(line_7) > 0 OR LENGTH(line_8) > 0)");
-            query.append(" ORDER BY rowid DESC");
+            query.append(" ORDER BY ").append(ConfigHandler.getDescendingEventOrder());
             filter.appendLimit(query);
 
             try (PreparedStatement statement = connection.prepareStatement(query.toString())) {

src/main/java/net/coreprotect/api/UsernameAPI.java

@@ -62,7 +62,7 @@ public class UsernameAPI {
                 appendPlaceholders(query, uuids.size());
                 query.append(")");
             }
-            query.append(" ORDER BY rowid DESC");
+            query.append(" ORDER BY ").append(ConfigHandler.getDescendingEventOrder());
             if (options.hasLimit()) {
                 query.append(" LIMIT ").append(options.getLimitCount()).append(" OFFSET ").append(options.getLimitOffset());
             }
@@ -137,7 +137,7 @@ public class UsernameAPI {
     private static UsernameResult parseUsernameResult(Connection connection, ResultSet results) throws Exception {
         String uuid = results.getString("uuid");
         String username = results.getString("user");
-        String player = UserStatement.getNameByUuid(uuid);
+        String player = UserStatement.getNameByUuid(connection, uuid);
         if (player == null) {
             player = username;
         }

src/main/java/net/coreprotect/command/lookup/StandardLookupThread.java

@@ -400,7 +400,7 @@ public class StandardLookupThread implements Runnable {
                         else if (actions.contains(LookupActions.USERNAME)) {
                             for (String[] data : lookupList) {
                                 String time = data[0];
-                                String user = UserStatement.getNameByUuid(data[1]);
+                                String user = UserStatement.getNameByUuid(connection, data[1]);
                                 String username = data[2];
                                 String timeago = ChatUtils.getTimeSince(Integer.parseInt(time), unixtimestamp, true);
                                 Chat.sendComponent(player, timeago + " " + Color.WHITE + "- " + Phrase.build(Phrase.LOOKUP_USERNAME, Color.DARK_AQUA + user + Color.WHITE, Color.DARK_AQUA + username + Color.WHITE));

src/main/java/net/coreprotect/command/parser/WorldParser.java

@@ -46,7 +46,7 @@ public class WorldParser {
                         world_id = WorldUtils.matchWorld(inputProcessed);
                         if (world_id == -1 && !requireLoaded) {
                             String worldName = argument.replaceFirst("#", "");
-                            world_id = ConfigHandler.worlds.getOrDefault(worldName, -1);
+                            world_id = ConfigHandler.resolveIdentifierId(ConfigHandler.CacheType.WORLDS, worldName, false);
                         }
                     }
                     next = 0;

src/main/java/net/coreprotect/config/Config.java

@@ -106,6 +106,7 @@ public class Config extends Language {
     public boolean WORLDEDIT;
     public int MAXIMUM_POOL_SIZE;
     public int CLICKHOUSE_PORT;
+    public int CLICKHOUSE_CONSUMER_DELAY;
     public int MYSQL_PORT;
     public int DEFAULT_RADIUS;
     public int DUCKDB_THREADS;
@@ -177,7 +178,7 @@ public class Config extends Language {
         HEADERS.put("donation-key", new String[] { "# CoreProtect is donationware. Obtain a donation key from coreprotect.net/donate/" });
         HEADERS.put("database-type", new String[] { "# Database engine used by CoreProtect. Valid values are duckdb, clickhouse, sqlite, and mysql.", "# Run /co reload or restart the server after changing the database engine or connection target." });
         HEADERS.put("mysql-host", new String[] { "# Connection settings for MySQL." });
-        HEADERS.put("clickhouse-host", new String[] { "# Connection settings for ClickHouse 25.6 or newer.", "# The configured database must already exist; CoreProtect creates its prefixed tables and views.", "# ClickHouse currently supports one CoreProtect installation per database and table prefix.", "# Preserve .clickhouse-writer with this installation and do not copy it to another active server." });
+        HEADERS.put("clickhouse-host", new String[] { "# Connection settings for ClickHouse 25.6 or newer.", "# The configured database must already exist; CoreProtect creates its prefixed tables and views.", "# Multiple writers require database-lock disabled, the same version and prefix, separate data directories, and direct connections to one physical server.", "# Keep writer clocks synchronized; a shared prefix is one logical namespace for worlds and players.", "# Do not reassign usernames, and record UUID-bearing logins before UUID-less activity under a changed name.", "# Stop every writer before migration or purge; purge requires database-lock on the remaining server.", "# Replicated, distributed, and load-balanced independent ClickHouse nodes are unsupported." });
         HEADERS.put("duckdb-memory-limit", new String[] { "# Resource limits for the embedded DuckDB database.", "# The memory limit controls DuckDB's buffer manager; the temporary limit caps spill data and is not preallocated." });
         HEADERS.put("language", new String[] { "# If modified, will automatically attempt to translate languages phrases.", "# List of language codes: https://coreprotect.net/languages/" });
         HEADERS.put("auto-purge", new String[] { "# Automatically purge data older than the configured time.", "# Examples: 30d, 12w, 6mo. Set to false to disable." });
@@ -241,6 +242,8 @@ public class Config extends Language {
         this.DATABASE_TYPE = this.getString("database-type");
         this.CLICKHOUSE_HOST = this.getString("clickhouse-host");
         this.CLICKHOUSE_PORT = this.getInt("clickhouse-port");
+        this.CLICKHOUSE_CONSUMER_DELAY = Math.max(500,
+                this.getInt("clickhouse-consumer-delay", 2500));
         this.CLICKHOUSE_DATABASE = this.getString("clickhouse-database");
         this.CLICKHOUSE_USERNAME = this.getString("clickhouse-username");
         this.CLICKHOUSE_PASSWORD = this.getString("clickhouse-password");

src/main/java/net/coreprotect/config/ConfigHandler.java

@@ -8,6 +8,7 @@ import java.sql.SQLException;
 import java.sql.Statement;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.EnumMap;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -17,9 +18,13 @@ import java.util.Set;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.IntConsumer;
+import java.util.function.IntSupplier;
+import java.util.function.Supplier;
 
 import org.bukkit.Bukkit;
 import org.bukkit.World;
+import org.bukkit.entity.EntityType;
 import org.bukkit.entity.Player;
 import org.bukkit.inventory.ItemStack;
 import org.bukkit.plugin.Plugin;
@@ -30,7 +35,9 @@ import com.zaxxer.hikari.HikariDataSource;
 
 import net.coreprotect.bukkit.BukkitAdapter;
 import net.coreprotect.consumer.Queue;
+import net.coreprotect.consumer.process.Process;
 import net.coreprotect.database.ConsumerWriteBatch;
+import net.coreprotect.database.ConsumerWriteBatch.ReferenceKind;
 import net.coreprotect.database.Database;
 import net.coreprotect.database.DatabaseType;
 import net.coreprotect.database.DuckDBNativeSupport;
@@ -49,6 +56,7 @@ import net.coreprotect.patch.Patch;
 import net.coreprotect.spigot.SpigotAdapter;
 import net.coreprotect.utility.Chat;
 import net.coreprotect.utility.Color;
+import net.coreprotect.utility.EntitySpawnTracking;
 import net.coreprotect.utility.ErrorReporter;
 import net.coreprotect.utility.SystemUtils;
 import net.coreprotect.utility.VersionUtils;
@@ -56,7 +64,17 @@ import net.coreprotect.utility.VersionUtils;
 public class ConfigHandler extends Queue {
 
     public enum CacheType {
-        MATERIALS, BLOCKDATA, ART, ENTITIES, WORLDS
+        MATERIALS(ReferenceKind.MATERIAL),
+        BLOCKDATA(ReferenceKind.BLOCK_DATA),
+        ART(ReferenceKind.ART),
+        ENTITIES(ReferenceKind.ENTITY),
+        WORLDS(ReferenceKind.WORLD);
+
+        private final ReferenceKind kind;
+
+        CacheType(ReferenceKind kind) {
+            this.kind = kind;
+        }
     }
 
     public static int SERVER_VERSION = 0;
@@ -84,6 +102,14 @@ public class ConfigHandler extends Queue {
     public static int maximumPoolSize = 10;
     public static DatabaseType databaseType = DatabaseType.SQLITE;
 
+    public static String getDescendingEventOrder() {
+        return databaseType.isClickHouse() ? "time DESC,rowid DESC" : "rowid DESC";
+    }
+
+    public static String getAscendingEventOrder() {
+        return databaseType.isClickHouse() ? "time ASC,rowid ASC" : "rowid ASC";
+    }
+
     public static final String BLACKLIST_COMMENT_SEPARATOR = ";";
     public static final String BLACKLIST_FILTER_SEPARATOR = "@";
     public static final String BLACKLIST_FILENAME = "blacklist.txt";
@@ -148,6 +174,51 @@ public class ConfigHandler extends Queue {
         }
     }
 
+    private static final class IdentifierStore {
+        private final ReferenceKind kind;
+        private final Supplier<Map<String, Integer>> values;
+        private final Supplier<Map<Integer, String>> reversed;
+        private final IntSupplier maximum;
+        private final IntConsumer maximumSetter;
+
+        private IdentifierStore(ReferenceKind kind, Supplier<Map<String, Integer>> values, Supplier<Map<Integer, String>> reversed, IntSupplier maximum,
+                IntConsumer maximumSetter) {
+            this.kind = kind;
+            this.values = values;
+            this.reversed = reversed;
+            this.maximum = maximum;
+            this.maximumSetter = maximumSetter;
+        }
+
+        private boolean store(int id, String value) {
+            if (!canStore(id, value)) {
+                return false;
+            }
+            Map<String, Integer> names = values.get();
+            Map<Integer, String> identifiers = reversed.get();
+            names.put(value, id);
+            identifiers.put(id, value);
+            maximumSetter.accept(Math.max(maximum.getAsInt(), id));
+            return true;
+        }
+
+        private boolean canStore(int id, String value) {
+            if (id < 1) {
+                return false;
+            }
+            Map<String, Integer> names = values.get();
+            Map<Integer, String> identifiers = reversed.get();
+            Integer storedId = names.get(value);
+            String storedValue = identifiers.get(id);
+            if ((storedId != null && storedId != id) || (storedValue != null && !storedValue.equals(value))) {
+                throw new IllegalStateException("CoreProtect " + kind.name().toLowerCase(Locale.ROOT) + " identifier " + id + "='" + value
+                        + "' conflicts with cached " + storedId + "='" + storedValue + "'");
+            }
+            return true;
+        }
+
+    }
+
     public static volatile Map<String, Integer> worlds = syncMap();
     public static volatile Map<Integer, String> worldsReversed = syncMap();
     public static volatile Map<String, Integer> materials = syncMap();
@@ -158,6 +229,7 @@ public class ConfigHandler extends Queue {
     public static volatile Map<Integer, String> entitiesReversed = syncMap();
     public static volatile Map<String, Integer> art = syncMap();
     public static volatile Map<Integer, String> artReversed = syncMap();
+    private static final Map<ReferenceKind, IdentifierStore> IDENTIFIER_STORES = identifierStores();
     public static Map<String, int[]> rollbackHash = syncMap();
     public static Map<String, Boolean> inspecting = syncMap();
     public static Map<String, Boolean> blacklist = syncMap();
@@ -220,8 +292,9 @@ public class ConfigHandler extends Queue {
             }
             try (ConsumerWriteBatch batch = Database.openConsumerWriteBatch(connection)) {
                 batch.begin();
+                int time = (int) (System.currentTimeMillis() / 1_000L);
                 for (Player player : Bukkit.getServer().getOnlinePlayers()) {
-                    batch.resolveUserId(player.getName(), player.getUniqueId().toString());
+                    batch.recordUsername(player.getName(), player.getUniqueId().toString(), 0, time);
                 }
                 if (!batch.commit()) {
                     throw new IllegalStateException("Unable to initialize ClickHouse player records");
@@ -391,9 +464,6 @@ public class ConfigHandler extends Queue {
     private static void initializeDatabase() {
         if (ConfigHandler.databaseType.isClickHouse()) {
             try {
-                if (!Config.getGlobal().DATABASE_LOCK) {
-                    throw new IllegalStateException("ClickHouse requires database-lock to remain enabled");
-                }
                 Config global = Config.getGlobal();
                 ClickHouseJdbcConfig config = new ClickHouseJdbcConfig(global.CLICKHOUSE_HOST, global.CLICKHOUSE_PORT,
                         global.CLICKHOUSE_DATABASE, global.CLICKHOUSE_USERNAME, global.CLICKHOUSE_PASSWORD,
@@ -551,12 +621,19 @@ public class ConfigHandler extends Queue {
         loaded &= loadBlockdata(statement);
         loaded &= loadArt(statement);
         loaded &= loadEntities(statement);
+        if (loaded && databaseType.isClickHouse() && !Config.getGlobal().DATABASE_LOCK) {
+            for (EntityType type : EntityType.values()) {
+                if (EntitySpawnTracking.isPlacedEntityType(type)) {
+                    resolveIdentifierId(CacheType.ENTITIES, type.name().toLowerCase(Locale.ROOT), true);
+                }
+            }
+        }
         return loaded;
     }
 
     /**
      * Reloads an identifier cache when multi-server cache refresh is permitted.
-     * 
+     *
      * @param type
      *             The type of cache to reload
      * @param name
@@ -564,46 +641,164 @@ public class ConfigHandler extends Queue {
      * @return The ID if found after reload, or -1 if not found
      */
     public static synchronized int reloadAndGetId(CacheType type, String name) {
-        if (Config.getGlobal().DATABASE_LOCK) {
+        if (Config.getGlobal().DATABASE_LOCK || !reloadIdentifierCache(type.kind)) {
+            return -1;
+        }
+        return identifierStore(type.kind).values.get().getOrDefault(name, -1);
+    }
+
+    public static int resolveIdentifierId(CacheType type, String value, boolean create) {
+        IdentifierStore store = identifierStore(type.kind);
+        Integer cached = store.values.get().get(value);
+        return cached != null ? cached : resolveMissingIdentifierId(store, value, create);
+    }
+
+    private static synchronized int resolveMissingIdentifierId(IdentifierStore store, String value, boolean create) {
+        Integer cached = store.values.get().get(value);
+        if (cached != null) {
+            return cached;
+        }
+
+        if (databaseType.isClickHouse()) {
+            int id;
+            try {
+                id = Database.findClickHouseIdentifierId(store.kind, value);
+            }
+            catch (Exception exception) {
+                throw new IllegalStateException("Unable to resolve ClickHouse " + store.kind.name().toLowerCase(Locale.ROOT) + " identifier for " + value, exception);
+            }
+            if (id > 0) {
+                store.store(id, value);
+                return id;
+            }
+        }
+        else if (create && !Config.getGlobal().DATABASE_LOCK && reloadIdentifierCache(store.kind)) {
+            cached = store.values.get().get(value);
+            if (cached != null) {
+                return cached;
+            }
+        }
+        if (!create) {
             return -1;
         }
 
-        if (!reloadIdentifierCache(type)) {
+        int id = allocateIdentifierId(store, value, store.maximum.getAsInt());
+        boolean clickHouse = databaseType.isClickHouse();
+        if (id == -1 || !store.canStore(id, value)) {
+            if (clickHouse) {
+                throw new IllegalStateException("Unable to resolve ClickHouse " + store.kind.name().toLowerCase(Locale.ROOT) + " identifier for " + value);
+            }
             return -1;
         }
+        if (clickHouse) {
+            if (!queueIdentifier(store.kind, id, value)) {
+                throw new IllegalStateException("Database persistence halted before the ClickHouse identifier could be queued");
+            }
+        }
+        store.store(id, value);
+        if (!clickHouse) {
+            queueIdentifier(store.kind, id, value);
+        }
+        return id;
+    }
 
-        switch (type) {
-            case MATERIALS:
-                return materials.getOrDefault(name, -1);
-            case BLOCKDATA:
-                return blockdata.getOrDefault(name, -1);
+    private static boolean queueIdentifier(ReferenceKind kind, int id, String value) {
+        switch (kind) {
             case ART:
-                return art.getOrDefault(name, -1);
-            case ENTITIES:
-                return entities.getOrDefault(name, -1);
-            case WORLDS:
-                return worlds.getOrDefault(name, -1);
+                return tryQueueIdentifierInsert(Process.ART_INSERT, id, value);
+            case BLOCK_DATA:
+                return tryQueueIdentifierInsert(Process.BLOCKDATA_INSERT, id, value);
+            case ENTITY:
+                return tryQueueIdentifierInsert(Process.ENTITY_INSERT, id, value);
+            case MATERIAL:
+                return tryQueueIdentifierInsert(Process.MATERIAL_INSERT, id, value);
+            case WORLD:
+                return tryQueueIdentifierInsert(Process.WORLD_INSERT, id, value);
             default:
-                return -1;
+                throw new IllegalArgumentException("Unsupported identifier cache: " + kind);
         }
     }
 
-    private static boolean reloadIdentifierCache(CacheType type) {
+    public static synchronized void uncacheIdentifier(ReferenceKind kind, int id, String value) {
+        IdentifierStore store = identifierStore(kind);
+        store.values.get().remove(value, Integer.valueOf(id));
+        store.reversed.get().remove(Integer.valueOf(id), value);
+    }
+
+    public static String getIdentifierValue(CacheType type, int id) {
+        IdentifierStore store = identifierStore(type.kind);
+        String value = store.reversed.get().get(id);
+        if (value != null || id < 1 || !databaseType.isClickHouse()) {
+            return value;
+        }
+        return loadMissingIdentifierValue(store, id);
+    }
+
+    private static synchronized String loadMissingIdentifierValue(IdentifierStore store, int id) {
+        String value = store.reversed.get().get(id);
+        if (value != null) {
+            return value;
+        }
+
+        try {
+            value = Database.findClickHouseIdentifierValue(store.kind, id);
+        }
+        catch (Exception exception) {
+            throw new IllegalStateException("Unable to resolve ClickHouse " + store.kind.name().toLowerCase(Locale.ROOT) + " identifier " + id, exception);
+        }
+        if (value != null) {
+            store.store(id, value);
+        }
+        return value;
+    }
+
+    private static int allocateIdentifierId(IdentifierStore store, String value, int currentMaximum) {
+        if (!ConfigHandler.databaseType.isClickHouse()) {
+            return currentMaximum + 1;
+        }
+
+        try {
+            return Database.nextClickHouseIdentifierId(store.kind, value, currentMaximum);
+        }
+        catch (Exception exception) {
+            throw new IllegalStateException("Unable to allocate ClickHouse " + store.kind.name().toLowerCase(Locale.ROOT) + " identifier for " + value, exception);
+        }
+    }
+
+    private static IdentifierStore identifierStore(ReferenceKind kind) {
+        IdentifierStore store = IDENTIFIER_STORES.get(kind);
+        if (store == null) {
+            throw new IllegalArgumentException("Unsupported identifier cache: " + kind);
+        }
+        return store;
+    }
+
+    private static Map<ReferenceKind, IdentifierStore> identifierStores() {
+        Map<ReferenceKind, IdentifierStore> stores = new EnumMap<>(ReferenceKind.class);
+        stores.put(ReferenceKind.MATERIAL, new IdentifierStore(ReferenceKind.MATERIAL, () -> materials, () -> materialsReversed, () -> materialId, id -> materialId = id));
+        stores.put(ReferenceKind.BLOCK_DATA, new IdentifierStore(ReferenceKind.BLOCK_DATA, () -> blockdata, () -> blockdataReversed, () -> blockdataId, id -> blockdataId = id));
+        stores.put(ReferenceKind.ART, new IdentifierStore(ReferenceKind.ART, () -> art, () -> artReversed, () -> artId, id -> artId = id));
+        stores.put(ReferenceKind.ENTITY, new IdentifierStore(ReferenceKind.ENTITY, () -> entities, () -> entitiesReversed, () -> entityId, id -> entityId = id));
+        stores.put(ReferenceKind.WORLD, new IdentifierStore(ReferenceKind.WORLD, () -> worlds, () -> worldsReversed, () -> worldId, id -> worldId = id));
+        return stores;
+    }
+
+    private static boolean reloadIdentifierCache(ReferenceKind kind) {
         try (Connection connection = Database.getConnection(true)) {
             if (connection == null) {
                 return false;
             }
             try (Statement statement = connection.createStatement()) {
-                switch (type) {
-                    case MATERIALS:
+                switch (kind) {
+                    case MATERIAL:
                         return loadMaterials(statement);
-                    case BLOCKDATA:
+                    case BLOCK_DATA:
                         return loadBlockdata(statement);
                     case ART:
                         return loadArt(statement);
-                    case ENTITIES:
+                    case ENTITY:
                         return loadEntities(statement);
-                    case WORLDS:
+                    case WORLD:
                         return loadWorlds(statement);
                     default:
                         return false;
@@ -623,25 +818,48 @@ public class ConfigHandler extends Queue {
             for (World world : worlds) {
                 String worldname = world.getName();
                 if (!cache.values.containsKey(worldname)) {
-                    int id = ++cache.maximumId;
+                    int id = allocateIdentifierId(identifierStore(ReferenceKind.WORLD), worldname, cache.maximumId);
+                    if (id == -1) {
+                        return false;
+                    }
+                    cache.maximumId = Math.max(cache.maximumId, id);
                     cache.values.put(worldname, id);
                     cache.reversed.put(id, worldname);
                     queuedWorlds.put(id, worldname);
                 }
             }
+            boolean clickHouse = databaseType.isClickHouse();
+            if (clickHouse) {
+                if (!queueIdentifiers(ReferenceKind.WORLD, queuedWorlds)) {
+                    throw new IllegalStateException("Database persistence halted before the ClickHouse world identifiers could be queued");
+                }
+            }
             ConfigHandler.worlds = syncMap(cache.values);
             ConfigHandler.worldsReversed = syncMap(cache.reversed);
             worldId = cache.maximumId;
-            for (Map.Entry<Integer, String> world : queuedWorlds.entrySet()) {
-                Queue.queueWorldInsert(world.getKey(), world.getValue());
+            if (!clickHouse) {
+                queueIdentifiers(ReferenceKind.WORLD, queuedWorlds);
             }
             return true;
-        } catch (Exception e) {
+        }
+        catch (Exception e) {
+            if (databaseType.isClickHouse() && e instanceof IllegalStateException) {
+                throw (IllegalStateException) e;
+            }
             ErrorReporter.report(e);
             return false;
         }
     }
 
+    private static boolean queueIdentifiers(ReferenceKind kind, Map<Integer, String> identifiers) {
+        for (Map.Entry<Integer, String> identifier : identifiers.entrySet()) {
+            if (!queueIdentifier(kind, identifier.getKey(), identifier.getValue())) {
+                return false;
+            }
+        }
+        return true;
+    }
+
     private static boolean checkDatabaseLock(Statement statement) {
         try {
             if (Config.getGlobal().DATABASE_LOCK) {

src/main/java/net/coreprotect/consumer/Consumer.java

@@ -443,6 +443,7 @@ public class Consumer extends Process implements Runnable, Thread.UncaughtExcept
     @Override
     public void run() {
         boolean lastRun = false;
+        boolean[] drained = { true, true };
 
         while (ConfigHandler.serverRunning || ConfigHandler.converterRunning || !lastRun) {
             if (!ConfigHandler.serverRunning && !ConfigHandler.converterRunning) {
@@ -465,15 +466,20 @@ public class Consumer extends Process implements Runnable, Thread.UncaughtExcept
                         currentConsumer = 0;
                     }
                 }
-                Thread.sleep(500);
+                Thread.sleep(consumerDelay(lastRun || !drained[0] || !drained[1]));
                 pauseConsumer(process_id);
                 databaseLifecycle.readLock().lock();
+                boolean processingAttempted = false;
                 try {
                     if (!databaseReloadPaused && !isPaused && !persistenceHalted) {
+                        processingAttempted = true;
                         Process.processConsumer(process_id, lastRun);
                     }
                 }
                 finally {
+                    if (processingAttempted) {
+                        drained[process_id] = getConsumerSize(process_id) == 0;
+                    }
                     databaseLifecycle.readLock().unlock();
                 }
             }

src/main/java/net/coreprotect/consumer/Queue.java

@@ -665,4 +665,8 @@ public class Queue {
     protected static void queueWorldInsert(int id, String world) {
         queueStandardData(new Object[] { null, Process.WORLD_INSERT, null, 0, null, 0, id, null }, new String[] { null, null }, world, false, Consumer.reserveConsumer());
     }
+
+    protected static boolean tryQueueIdentifierInsert(int action, int id, String value) {
+        return queueStandardData(new Object[] { null, action, null, 0, null, 0, id, null }, new String[] { null, null }, value, false, Consumer.reserveConsumer());
+    }
 }

src/main/java/net/coreprotect/consumer/process/Process.java

@@ -15,6 +15,7 @@ import java.util.UUID;
 import org.bukkit.Location;
 import org.bukkit.Material;
 
+import net.coreprotect.config.Config;
 import net.coreprotect.config.ConfigHandler;
 import net.coreprotect.consumer.Consumer;
 import net.coreprotect.consumer.Queue;
@@ -82,6 +83,13 @@ public class Process {
         return currentConsumerSize;
     }
 
+    protected static int consumerDelay(boolean backlog) {
+        if (backlog || !ConfigHandler.databaseType.isClickHouse()) {
+            return 500;
+        }
+        return Math.max(500, Config.getGlobal().CLICKHOUSE_CONSUMER_DELAY);
+    }
+
     public static boolean isRollbackPublication(int action, Object object) {
         if (action == ROLLBACK_UPDATE || action == CONTAINER_ROLLBACK_UPDATE || action == INVENTORY_ROLLBACK_UPDATE || action == INVENTORY_CONTAINER_ROLLBACK_UPDATE || action == BLOCK_INVENTORY_ROLLBACK_UPDATE || action == ENTITY_CONTAINER_ROLLBACK_UPDATE || action == ENTITY_CONTAINER_TRANSITION_UPDATE) {
             return true;
@@ -472,9 +480,10 @@ public class Process {
                                     break;
                             }
 
-                            // If interrupt requested, commit data, sleep, and resume processing
+                            // Commit full or interrupted batches before continuing.
                             boolean interrupted = Consumer.interrupt;
-                            if ((interrupted || writeBatch.shouldCommit()) && !isolatedTransaction) {
+                            boolean batchLimitReached = writeBatch.shouldCommit();
+                            if ((interrupted || batchLimitReached) && !isolatedTransaction) {
                                 boolean committed = commit(writeBatch);
                                 if (committed) {
                                     processedThrough = i + 1;
@@ -484,9 +493,10 @@ public class Process {
                                     failConsumerBatch(processId, consumerData, users, consumerObject, processedThrough, i + 1);
                                     return;
                                 }
-                                if (interrupted) {
+                                boolean backlog = batchLimitReached && ConfigHandler.databaseType.isClickHouse() && i + 1 < consumerDataSize;
+                                if (interrupted || backlog) {
                                     try {
-                                        Thread.sleep(500);
+                                        Thread.sleep(consumerDelay(true));
                                     }
                                     catch (InterruptedException exception) {
                                         Thread.currentThread().interrupt();

src/main/java/net/coreprotect/consumer/process/SignUpdateProcess.java

@@ -27,10 +27,10 @@ class SignUpdateProcess {
             int userid = ConfigHandler.playerIdCache.get(user.toLowerCase(Locale.ROOT));
             String query = "";
             if (action == 0) {
-                query = "SELECT color, color_secondary, data, waxed, face, line_1, line_2, line_3, line_4, line_5, line_6, line_7, line_8 FROM " + ConfigHandler.prefix + "sign WHERE " + ConfigHandler.databaseType.getUserColumn() + "=" + userid + " AND wid=" + wid + " AND x=" + x + " AND z=" + z + " AND y=" + y + " AND time < " + time + " ORDER BY rowid DESC LIMIT 1 OFFSET 0";
+                query = "SELECT color, color_secondary, data, waxed, face, line_1, line_2, line_3, line_4, line_5, line_6, line_7, line_8 FROM " + ConfigHandler.prefix + "sign WHERE " + ConfigHandler.databaseType.getUserColumn() + "=" + userid + " AND wid=" + wid + " AND x=" + x + " AND z=" + z + " AND y=" + y + " AND time < " + time + " ORDER BY " + ConfigHandler.getDescendingEventOrder() + " LIMIT 1 OFFSET 0";
             }
             else {
-                query = "SELECT color, color_secondary, data, waxed, face, line_1, line_2, line_3, line_4, line_5, line_6, line_7, line_8 FROM " + ConfigHandler.prefix + "sign WHERE " + ConfigHandler.databaseType.getUserColumn() + "=" + userid + " AND wid=" + wid + " AND x=" + x + " AND z=" + z + " AND y=" + y + " AND time >= " + time + " ORDER BY rowid ASC LIMIT 1 OFFSET 0";
+                query = "SELECT color, color_secondary, data, waxed, face, line_1, line_2, line_3, line_4, line_5, line_6, line_7, line_8 FROM " + ConfigHandler.prefix + "sign WHERE " + ConfigHandler.databaseType.getUserColumn() + "=" + userid + " AND wid=" + wid + " AND x=" + x + " AND z=" + z + " AND y=" + y + " AND time >= " + time + " ORDER BY " + ConfigHandler.getAscendingEventOrder() + " LIMIT 1 OFFSET 0";
             }
             SignStatement.getData(statement, block, query);
             BlockUtils.updateBlock(block);

src/main/java/net/coreprotect/database/BlockLookup.java

@@ -31,7 +31,7 @@ public class BlockLookup {
             int worldId = WorldUtils.getWorldId(block.getWorld().getName());
             String table = DuckDBLookupQuery.spatialTable(statement.getConnection(), "block", worldId, x, x, z, z, "spatial_rows");
             String index = ConfigHandler.databaseType.isDuckDB() ? "" : WorldUtils.getWidIndex("block");
-            String query = "SELECT " + ConfigHandler.databaseType.getUserColumn() + ",type FROM " + table + " " + index + "WHERE wid = " + worldId + " AND x = " + x + " AND z = " + z + " AND y = " + y + " AND rolled_back IN(0,2) AND action=1 ORDER BY rowid DESC LIMIT 1 OFFSET 0";
+            String query = "SELECT " + ConfigHandler.databaseType.getUserColumn() + ",type FROM " + table + " " + index + "WHERE wid = " + worldId + " AND x = " + x + " AND z = " + z + " AND y = " + y + " AND rolled_back IN(0,2) AND action=1 ORDER BY " + ConfigHandler.getDescendingEventOrder() + " LIMIT 1 OFFSET 0";
 
             ResultSet results = statement.executeQuery(query);
             while (results.next()) {

src/main/java/net/coreprotect/database/Database.java

@@ -30,6 +30,7 @@ import net.coreprotect.consumer.process.Process;
 import net.coreprotect.database.clickhouse.ClickHouseConsumerWriteBatch;
 import net.coreprotect.database.clickhouse.ClickHouseDatabase;
 import net.coreprotect.database.clickhouse.ClickHouseJdbcConfig;
+import net.coreprotect.database.statement.UserStatement;
 import net.coreprotect.language.Phrase;
 import net.coreprotect.listener.player.InventoryChangeListener;
 import net.coreprotect.model.BlockGroup;
@@ -598,6 +599,18 @@ public class Database extends Queue {
         return new RelationalConsumerWriteBatch(connection, ConfigHandler.databaseType);
     }
 
+    public static int nextClickHouseIdentifierId(ConsumerWriteBatch.ReferenceKind kind, String value, int currentMaximum) throws SQLException {
+        return requireClickHouseDatabase().nextIdentifierId(kind, value, currentMaximum);
+    }
+
+    public static String findClickHouseIdentifierValue(ConsumerWriteBatch.ReferenceKind kind, int id) throws SQLException {
+        return requireClickHouseDatabase().findIdentifierValue(kind, id);
+    }
+
+    public static int findClickHouseIdentifierId(ConsumerWriteBatch.ReferenceKind kind, String value) throws SQLException {
+        return requireClickHouseDatabase().findIdentifierId(kind, value);
+    }
+
     public static void recordDatabaseVersion(Statement statement, String version) throws SQLException {
         if (ConfigHandler.databaseType.isClickHouse()) {
             requireClickHouseDatabase().updateCoreVersion(version);
@@ -608,6 +621,9 @@ public class Database extends Queue {
     }
 
     public static long purgeClickHouse(long startTime, long endTime, int worldId, List<Integer> blockTypes, boolean optimize) throws SQLException {
+        if (!Config.getGlobal().DATABASE_LOCK) {
+            throw new SQLException("ClickHouse purge requires database-lock to be enabled and every other CoreProtect installation sharing the database and prefix to be stopped");
+        }
         return requireClickHouseDatabase().purge(startTime, endTime, worldId, blockTypes, optimize);
     }
 
@@ -1186,6 +1202,7 @@ public class Database extends Queue {
     private static synchronized void closeClickHouseChecked() throws SQLException {
         ClickHouseDatabase database = clickHouseDatabase;
         clickHouseDatabase = null;
+        UserStatement.clearClickHouseCaches();
         if (database != null) {
             database.close();
         }

src/main/java/net/coreprotect/database/LookupRaw.java

@@ -923,7 +923,7 @@ public class LookupRaw extends Queue {
             }
 
             String rows = summary ? "rowid as id,time," + userColumn + ",wid,x,y,z,type,meta as metadata,data,-1 as amount,action,rolled_back,0 as entity_spawn_rowid" : "rowid as id,time," + userColumn + ",wid,x,y,z,action,type,data,meta,blockdata,rolled_back";
-            String queryOrder = " ORDER BY rowid DESC";
+            String queryOrder = " ORDER BY " + ConfigHandler.getDescendingEventOrder();
 
             if (actionList.contains(LookupActions.CONTAINER) || actionList.contains(5)) {
                 queryTable = "container";

src/main/java/net/coreprotect/database/clickhouse/ClickHouseBatchPublisher.java

@@ -4,6 +4,7 @@ import java.sql.Connection;
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
+import java.util.Map;
 import java.util.Objects;
 
 import net.coreprotect.config.ConfigHandler;
@@ -40,17 +41,29 @@ final class ClickHouseBatchPublisher {
             return receipt;
         }
         if (receipt.getRowCount() > MAX_INSERT_ROWS) {
-            throw new SQLException("ClickHouse batches cannot exceed " + MAX_INSERT_ROWS + " physical rows");
+            throw new SQLException("ClickHouse batches cannot exceed " + MAX_INSERT_ROWS + " data rows");
         }
+        for (Map.Entry<Integer, Integer> partition : receipt.getPartitionRowCounts().entrySet()) {
+            if (partition.getValue() >= MAX_INSERT_ROWS) {
+                throw new SQLException("ClickHouse batch partition " + partition.getKey() + " cannot exceed " + (MAX_INSERT_ROWS - 1) + " data rows");
+            }
+            if (!batch.isPartitionPublished(partition.getKey())) {
+                publishPartition(batch, receipt, partition.getKey(), partition.getValue());
+            }
+        }
+        batch.markPublished();
+        return receipt;
+    }
 
+    private void publishPartition(ClickHouseWriteBatch batch, ClickHouseBatchReceipt receipt, int partitionId, int expectedRowCount) throws SQLException {
         SQLException publicationFailure = null;
         int attempt = 1;
         while (true) {
             try {
                 writerRegistration.verifyOwned();
-                nativeClient.insert(eventTable, ClickHouseSchema.EVENT_COLUMNS, batch.openRows(), batch.getIdentity(), "records");
-                batch.markPublished();
-                return receipt;
+                nativeClient.insert(eventTable, ClickHouseSchema.EVENT_COLUMNS, batch.openRows(partitionId), batch.getIdentity(), "records_" + partitionId);
+                batch.markPartitionPublished(partitionId);
+                return;
             }
             catch (ClickHouseWriterRegistration.OwnershipException exception) {
                 throw exception;
@@ -59,69 +72,78 @@ final class ClickHouseBatchPublisher {
                 publicationFailure = exception;
             }
 
-            RawStatus status;
+            ReceiptStatus status;
             try {
-                status = readStatus(batch.getIdentity());
+                status = readStatus(batch.getIdentity(), partitionId);
             }
             catch (SQLException reconciliationFailure) {
                 reconciliationFailure.addSuppressed(publicationFailure);
                 throw reconciliationFailure;
             }
-            if (status.matches(receipt)) {
-                batch.markPublished();
-                return receipt;
+            if (status.matches(receipt, expectedRowCount)) {
+                batch.markPartitionPublished(partitionId);
+                return;
             }
-            if (!status.isEmpty() && !status.isRetryablePartial(receipt)) {
-                SQLException conflict = conflict(receipt);
+            if (!status.isEmpty()) {
+                SQLException conflict = conflict(receipt, partitionId);
                 conflict.addSuppressed(publicationFailure);
                 throw conflict;
             }
             if (attempt >= MAX_PUBLICATION_ATTEMPTS && !shouldContinueRecovery()) {
-                throw new SQLException("ClickHouse batch remained incomplete after " + attempt + " publication attempts", publicationFailure);
+                throw new SQLException("ClickHouse batch partition " + partitionId + " remained incomplete after " + attempt + " publication attempts", publicationFailure);
+            }
+            pauseBeforeRetry(attempt, "republishing a ClickHouse batch partition");
+            if (attempt < Integer.MAX_VALUE) {
+                attempt++;
             }
-            pauseBeforeRetry(attempt++, "republishing a ClickHouse batch");
         }
     }
 
-    private RawStatus readStatus(ClickHouseBatchIdentity identity) throws SQLException {
+    private ReceiptStatus readStatus(ClickHouseBatchIdentity identity, int partitionId) throws SQLException {
         SQLException failure = null;
-        for (int attempt = 1; attempt <= MAX_RECONCILIATION_ATTEMPTS; attempt++) {
+        int attempt = 1;
+        while (true) {
             if (Thread.currentThread().isInterrupted()) {
                 throw interrupted("reconciling a ClickHouse batch", failure);
             }
             try {
-                return readRawStatus(identity);
+                return readReceiptStatus(identity, partitionId);
             }
             catch (SQLException exception) {
-                if (failure != null && failure != exception) {
-                    exception.addSuppressed(failure);
-                }
                 failure = exception;
             }
-            if (attempt < MAX_RECONCILIATION_ATTEMPTS) {
-                pauseBeforeRetry(attempt, "reconciling a ClickHouse batch");
+            if (attempt >= MAX_RECONCILIATION_ATTEMPTS && !shouldContinueRecovery()) {
+                throw new SQLException("ClickHouse batch status remained unavailable after " + attempt + " attempts", failure);
+            }
+            pauseBeforeRetry(attempt, "reconciling a ClickHouse batch");
+            if (attempt < Integer.MAX_VALUE) {
+                attempt++;
             }
         }
-        throw new SQLException("ClickHouse batch status remained unavailable after " + MAX_RECONCILIATION_ATTEMPTS + " attempts", failure);
     }
 
-    private RawStatus readRawStatus(ClickHouseBatchIdentity identity) throws SQLException {
-        String sql = "SELECT count(),uniqExact(tuple(family,rowid)),uniqExact(batch_id),any(toString(batch_id)),uniqExact(batch_ordinal),max(batch_ordinal)"
+    private ReceiptStatus readReceiptStatus(ClickHouseBatchIdentity identity, int partitionId) throws SQLException {
+        String sql = "SELECT toString(batch_id),ifNull(amount,-1)"
                 + " FROM " + eventTable
-                + " WHERE batch_sequence=?";
+                + " WHERE family=? AND batch_sequence=? AND rowid=? AND wid=?"
+                + " GROUP BY batch_id,amount LIMIT 2";
         try (Connection connection = jdbc.openConnection(); PreparedStatement statement = connection.prepareStatement(sql)) {
-            statement.setLong(1, identity.getBatchSequence());
+            statement.setString(1, ClickHouseSchema.BATCH_RECEIPT_FAMILY);
+            statement.setLong(2, identity.getBatchSequence());
+            statement.setLong(3, identity.getBatchSequence());
+            statement.setInt(4, partitionId);
             try (ResultSet resultSet = statement.executeQuery()) {
                 if (!resultSet.next()) {
-                    throw new SQLException("ClickHouse did not return batch reconciliation data");
+                    return ReceiptStatus.EMPTY;
                 }
-                return new RawStatus(resultSet.getLong(1), resultSet.getLong(2), resultSet.getLong(3), resultSet.getString(4), resultSet.getLong(5), resultSet.getLong(6));
+                ReceiptStatus status = new ReceiptStatus(resultSet.getString(1), resultSet.getInt(2), false);
+                return resultSet.next() ? ReceiptStatus.CONFLICTING : status;
             }
         }
     }
 
-    private static SQLException conflict(ClickHouseBatchReceipt receipt) {
-        return new SQLException("ClickHouse batch is partial or conflicting for batch sequence " + receipt.getBatchSequence());
+    private static SQLException conflict(ClickHouseBatchReceipt receipt, int partitionId) {
+        return new SQLException("ClickHouse batch receipt is conflicting for batch sequence " + receipt.getBatchSequence() + " partition " + partitionId);
     }
 
     static SQLException interrupted(String operation, SQLException failure) {
@@ -148,47 +170,28 @@ final class ClickHouseBatchPublisher {
         return ConfigHandler.serverRunning || ConfigHandler.shutdownDrainRunning;
     }
 
-    private static final class RawStatus {
+    private static final class ReceiptStatus {
+
+        private static final ReceiptStatus EMPTY = new ReceiptStatus(null, -1, false);
+        private static final ReceiptStatus CONFLICTING = new ReceiptStatus(null, -1, true);
 
-        private final long count;
-        private final long logicalRows;
-        private final long uniqueBatchIds;
         private final String batchId;
-        private final long uniqueOrdinals;
-        private final long maximumOrdinal;
+        private final int rowCount;
+        private final boolean conflicting;
 
-        private RawStatus(long count, long logicalRows, long uniqueBatchIds, String batchId, long uniqueOrdinals, long maximumOrdinal) {
-            this.count = count;
-            this.logicalRows = logicalRows;
-            this.uniqueBatchIds = uniqueBatchIds;
+        private ReceiptStatus(String batchId, int rowCount, boolean conflicting) {
             this.batchId = batchId;
-            this.uniqueOrdinals = uniqueOrdinals;
-            this.maximumOrdinal = maximumOrdinal;
+            this.rowCount = rowCount;
+            this.conflicting = conflicting;
         }
 
         private boolean isEmpty() {
-            return count == 0;
+            return this == EMPTY;
         }
 
-        private boolean matches(ClickHouseBatchReceipt receipt) {
-            long expectedLogicalRows = receipt.getLogicalRowCount();
-            return count >= expectedLogicalRows
-                    && count <= receipt.getRowCount()
-                    && logicalRows == expectedLogicalRows
-                    && uniqueBatchIds == 1
-                    && uniqueOrdinals == count
-                    && maximumOrdinal == receipt.getRowCount() - 1L
-                    && receipt.getBatchId().toString().equals(batchId);
-        }
-
-        private boolean isRetryablePartial(ClickHouseBatchReceipt receipt) {
-            return count > 0
-                    && count < receipt.getRowCount()
-                    && logicalRows > 0
-                    && logicalRows <= receipt.getLogicalRowCount()
-                    && uniqueBatchIds == 1
-                    && uniqueOrdinals == count
-                    && maximumOrdinal < receipt.getRowCount()
+        private boolean matches(ClickHouseBatchReceipt receipt, int expectedRowCount) {
+            return !conflicting
+                    && rowCount == expectedRowCount
                     && receipt.getBatchId().toString().equals(batchId);
         }
     }

src/main/java/net/coreprotect/database/clickhouse/ClickHouseBatchReceipt.java

@@ -1,6 +1,9 @@
 package net.coreprotect.database.clickhouse;
 
+import java.util.Collections;
+import java.util.Map;
 import java.util.Objects;
+import java.util.TreeMap;
 import java.util.UUID;
 
 public final class ClickHouseBatchReceipt {
@@ -9,15 +12,28 @@ public final class ClickHouseBatchReceipt {
     private final UUID batchId;
     private final int rowCount;
     private final int logicalRowCount;
+    private final Map<Integer, Integer> partitionRowCounts;
 
-    ClickHouseBatchReceipt(long batchSequence, UUID batchId, int rowCount, int logicalRowCount) {
+    ClickHouseBatchReceipt(long batchSequence, UUID batchId, int rowCount, int logicalRowCount, Map<Integer, Integer> partitionRowCounts) {
         if (batchSequence < 1 || rowCount < 0 || logicalRowCount < 0 || logicalRowCount > rowCount) {
             throw new IllegalArgumentException("Invalid ClickHouse batch receipt");
         }
+        TreeMap<Integer, Integer> partitions = new TreeMap<>(Objects.requireNonNull(partitionRowCounts, "partitionRowCounts"));
+        int partitionRowCount = 0;
+        for (Map.Entry<Integer, Integer> partition : partitions.entrySet()) {
+            if (partition.getKey() < 0 || partition.getValue() == null || partition.getValue() < 1) {
+                throw new IllegalArgumentException("Invalid ClickHouse batch partition receipt");
+            }
+            partitionRowCount = Math.addExact(partitionRowCount, partition.getValue());
+        }
+        if (partitionRowCount != rowCount) {
+            throw new IllegalArgumentException("ClickHouse batch partition counts do not match the row count");
+        }
         this.batchSequence = batchSequence;
         this.batchId = Objects.requireNonNull(batchId, "batchId");
         this.rowCount = rowCount;
         this.logicalRowCount = logicalRowCount;
+        this.partitionRowCounts = Collections.unmodifiableMap(partitions);
     }
 
     public long getBatchSequence() {
@@ -36,4 +52,8 @@ public final class ClickHouseBatchReceipt {
         return logicalRowCount;
     }
 
+    Map<Integer, Integer> getPartitionRowCounts() {
+        return partitionRowCounts;
+    }
+
 }

src/main/java/net/coreprotect/database/clickhouse/ClickHouseConsumerWriteBatch.java

@@ -5,6 +5,7 @@ import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.EnumMap;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -30,10 +31,8 @@ public final class ClickHouseConsumerWriteBatch implements ConsumerWriteBatch {
     private static final int TARGET_BATCH_ROWS = 100_000;
 
     private final ClickHouseDatabase database;
-    private final String tablePrefix;
     private final String userTable;
     private final String usernameLogTable;
-    private final String databaseLockTable;
     private final ClickHouseEntitySpawnUpdates entitySpawnUpdates;
     private final EnumMap<ClickHouseFamily, Map<Long, ClickHouseEventPointer>> localPointers = new EnumMap<>(ClickHouseFamily.class);
     private final Map<Long, Integer> localBlockActions = new HashMap<>();
@@ -42,17 +41,15 @@ public final class ClickHouseConsumerWriteBatch implements ConsumerWriteBatch {
     private final Map<String, Integer> userAliases = new HashMap<>();
     private final Set<String> usernameHistory = new HashSet<>();
     private final EnumMap<ReferenceKind, Map<Integer, String>> localReferences = new EnumMap<>(ReferenceKind.class);
+    private final EnumMap<ReferenceKind, Map<Integer, String>> unpublishedReferences = new EnumMap<>(ReferenceKind.class);
     private ClickHouseWriteBatch batch;
-    private Long databaseLockRowId;
     private boolean closed;
 
     public ClickHouseConsumerWriteBatch(ClickHouseDatabase database, String prefix) {
         this.database = Objects.requireNonNull(database, "database");
         String validatedPrefix = prefix == null || prefix.isEmpty() ? "" : ClickHouseIdentifiers.requireIdentifier(prefix, "ClickHouse table prefix");
-        tablePrefix = validatedPrefix;
         userTable = validatedPrefix + ClickHouseFamily.USER.getTableName();
         usernameLogTable = validatedPrefix + ClickHouseFamily.USERNAME_LOG.getTableName();
-        databaseLockTable = validatedPrefix + ClickHouseFamily.DATABASE_LOCK.getTableName();
         entitySpawnUpdates = new ClickHouseEntitySpawnUpdates(this, validatedPrefix + "event_data");
     }
 
@@ -90,12 +87,13 @@ public final class ClickHouseConsumerWriteBatch implements ConsumerWriteBatch {
         finally {
             current.close();
             batch = null;
+            if (!published) {
+                discardReferenceCache(Collections.emptyMap());
+                entitySpawnUpdates.batchDiscarded();
+            }
             clearBatchState();
             Consumer.transacting = false;
             Consumer.interrupt = false;
-            if (!published) {
-                entitySpawnUpdates.batchDiscarded();
-            }
         }
     }
 
@@ -111,6 +109,7 @@ public final class ClickHouseConsumerWriteBatch implements ConsumerWriteBatch {
         }
         batch.close();
         batch = null;
+        discardReferenceCache(Collections.emptyMap());
         clearBatchState();
         entitySpawnUpdates.batchDiscarded();
         Consumer.transacting = false;
@@ -148,21 +147,51 @@ public final class ClickHouseConsumerWriteBatch implements ConsumerWriteBatch {
         String requiredUser = requireText(user, "user");
         String normalizedUuid = normalizeUuid(uuid);
         String cacheKey = normalizeName(requiredUser);
-        Integer cachedId = ConfigHandler.playerIdCache.get(cacheKey);
+        Integer cachedId = cachedUserId(cacheKey, normalizedUuid);
         if (cachedId != null) {
             return cachedId;
         }
-        UserRow row = findStagedUser(requiredUser, normalizedUuid);
-        if (row == null) {
-            row = findUser(requiredUser, normalizedUuid);
+        return resolveUser(requiredUser, normalizedUuid, currentTime(), false).row.id;
+    }
+
+    private UserResolution resolveUser(String user, String uuid, int time, boolean updateIdentity) throws Exception {
+        String cacheKey = normalizeName(user);
+        UserRow row = null;
+        if (!uuid.isEmpty()) {
+            row = usersByUuid.get(uuid);
+            if (row == null) {
+                row = findUserByUuid(uuid);
+            }
         }
         if (row == null) {
-            long rowId = events().addUser(currentTime(), requiredUser, normalizedUuid);
-            row = new UserRow(toIntId(rowId, "user"), requiredUser, normalizedUuid);
+            row = usersByName.get(cacheKey);
+            if (row == null) {
+                row = findUserByName(user);
+            }
+        }
+
+        int rowId = database.canonicalUserId(user, uuid, row == null ? null : row.id);
+        if (row == null || row.id != rowId) {
+            row = findStagedUserById(rowId);
+            if (row == null) {
+                row = findUserById(rowId);
+            }
+        }
+        boolean changed = row == null;
+        if (changed) {
+            events().addUserVersion(rowId, time, user, uuid);
+            row = new UserRow(rowId, user, uuid);
+        }
+        else if (!uuid.isEmpty() && (row.uuid.isEmpty()
+                || (updateIdentity && (!user.equalsIgnoreCase(row.name) || !uuid.equals(row.uuid))))) {
+            events().addUserVersion(rowId, time, user, uuid);
+            unstageUser(row);
+            row = new UserRow(rowId, user, uuid);
+            changed = true;
         }
         stageUser(row);
         userAliases.put(cacheKey, row.id);
-        return row.id;
+        return new UserResolution(row, changed);
     }
 
     @Override
@@ -173,33 +202,8 @@ public final class ClickHouseConsumerWriteBatch implements ConsumerWriteBatch {
         if (ConfigHandler.isBlacklisted(requiredUser)) {
             return;
         }
-        UserRow row = usersByUuid.get(requiredUuid);
-        if (row == null) {
-            row = findUserByUuid(requiredUuid);
-        }
-        if (row == null) {
-            row = findStagedUser(requiredUser, requiredUuid);
-        }
-        if (row == null) {
-            row = findUser(requiredUser, requiredUuid);
-        }
-        if (row == null) {
-            int userId = resolveUserId(requiredUser, requiredUuid);
-            row = usersByName.get(normalizeName(requiredUser));
-            if (row == null || row.id != userId) {
-                throw new SQLException("Unable to stage ClickHouse user " + requiredUser);
-            }
-        }
-
-        boolean changed = !row.name.equalsIgnoreCase(requiredUser) || !requiredUuid.equals(row.uuid);
-        if (changed) {
-            events().addUserVersion(row.id, time, requiredUser, requiredUuid);
-            unstageUser(row);
-            row = new UserRow(row.id, requiredUser, requiredUuid);
-            stageUser(row);
-        }
-
-        if (retainHistory == 1 && (changed || !hasUsernameHistory(requiredUuid, requiredUser))) {
+        UserResolution resolution = resolveUser(requiredUser, requiredUuid, time, true);
+        if (retainHistory == 1 && (resolution.changed || !hasUsernameHistory(requiredUuid, requiredUser))) {
             events().addUsernameLog(time, requiredUuid, requiredUser);
             usernameHistory.add(historyKey(requiredUuid, requiredUser));
         }
@@ -208,15 +212,7 @@ public final class ClickHouseConsumerWriteBatch implements ConsumerWriteBatch {
     @Override
     public void updateDatabaseLock(int status, int time) throws Exception {
         requireBatch();
-        if (databaseLockRowId == null) {
-            databaseLockRowId = findDatabaseLockRowId();
-        }
-        if (databaseLockRowId == null) {
-            databaseLockRowId = events().addDatabaseLock(time, status);
-        }
-        else {
-            events().addDatabaseLockVersion(databaseLockRowId, time, status);
-        }
+        events().addDatabaseLock(time, status);
     }
 
     @Override
@@ -237,11 +233,9 @@ public final class ClickHouseConsumerWriteBatch implements ConsumerWriteBatch {
             }
             return;
         }
-        if (hasCommittedReference(kind, id, requiredValue)) {
-            references.put(id, requiredValue);
-            return;
-        }
 
+        Map<Integer, String> unpublished = unpublishedReferences.computeIfAbsent(kind, ignored -> new HashMap<>());
+        unpublished.put(id, requiredValue);
         references.put(id, requiredValue);
         try {
             switch (kind) {
@@ -430,7 +424,7 @@ public final class ClickHouseConsumerWriteBatch implements ConsumerWriteBatch {
     }
 
     Connection openConnection() throws SQLException {
-        return database.openConnection();
+        return database.openAuxiliaryConnection();
     }
 
     ClickHouseDatabase database() {
@@ -494,37 +488,56 @@ public final class ClickHouseConsumerWriteBatch implements ConsumerWriteBatch {
         return pointer;
     }
 
-    private UserRow findStagedUser(String user, String uuid) {
-        UserRow row = usersByName.get(normalizeName(user));
-        if (row == null && !uuid.isEmpty()) {
-            row = usersByUuid.get(uuid);
+    private UserRow findStagedUserById(int id) {
+        for (UserRow row : usersByName.values()) {
+            if (row.id == id) {
+                return row;
+            }
         }
-        return row;
+        for (UserRow row : usersByUuid.values()) {
+            if (row.id == id) {
+                return row;
+            }
+        }
+        return null;
     }
 
-    private UserRow findUser(String user, String uuid) throws SQLException {
-        String sql = "SELECT rowid,`user`,uuid FROM " + userTable + " WHERE lowerUTF8(`user`)=lowerUTF8(?)";
-        if (!uuid.isEmpty()) {
-            sql += " OR uuid=?";
+    private Integer cachedUserId(String user, String uuid) {
+        Integer id = userAliases.get(user);
+        if (id != null) {
+            UserRow row = findStagedUserById(id);
+            return uuid.isEmpty() || (row != null && uuid.equals(row.uuid)) ? id : null;
         }
-        sql += " ORDER BY rowid LIMIT 1";
+        id = ConfigHandler.playerIdCache.get(user);
+        return id != null && (uuid.isEmpty() || uuid.equals(ConfigHandler.uuidCache.get(user))) ? id : null;
+    }
+
+    private UserRow findUserByName(String user) throws SQLException {
+        String sql = "SELECT rowid,`user`,uuid FROM " + userTable
+                + " WHERE lowerUTF8(`user`)=lowerUTF8(?) ORDER BY (uuid!='') DESC,rowid LIMIT 1";
         try (Connection connection = openConnection(); PreparedStatement statement = connection.prepareStatement(sql)) {
             statement.setString(1, user);
-            if (!uuid.isEmpty()) {
-                statement.setString(2, uuid);
-            }
-            return readUniqueUser(statement);
+            return readUser(statement);
         }
     }
 
     private UserRow findUserByUuid(String uuid) throws SQLException {
-        try (Connection connection = openConnection(); PreparedStatement statement = connection.prepareStatement("SELECT rowid,`user`,uuid FROM " + userTable + " WHERE uuid=? ORDER BY rowid LIMIT 1")) {
+        try (Connection connection = openConnection(); PreparedStatement statement = connection.prepareStatement(
+                "SELECT rowid,`user`,uuid FROM " + userTable
+                        + " WHERE uuid=? ORDER BY rowid LIMIT 1")) {
             statement.setString(1, uuid);
-            return readUniqueUser(statement);
+            return readUser(statement);
         }
     }
 
-    private UserRow readUniqueUser(PreparedStatement statement) throws SQLException {
+    private UserRow findUserById(int id) throws SQLException {
+        try (Connection connection = openConnection(); PreparedStatement statement = connection.prepareStatement("SELECT rowid,`user`,uuid FROM " + userTable + " WHERE rowid=? LIMIT 1")) {
+            statement.setInt(1, id);
+            return readUser(statement);
+        }
+    }
+
+    private UserRow readUser(PreparedStatement statement) throws SQLException {
         try (ResultSet resultSet = statement.executeQuery()) {
             if (!resultSet.next()) {
                 return null;
@@ -553,42 +566,7 @@ public final class ClickHouseConsumerWriteBatch implements ConsumerWriteBatch {
         return false;
     }
 
-    private Long findDatabaseLockRowId() throws SQLException {
-        try (Connection connection = openConnection(); PreparedStatement statement = connection.prepareStatement("SELECT rowid FROM " + databaseLockTable + " ORDER BY rowid LIMIT 2"); ResultSet resultSet = statement.executeQuery()) {
-            if (!resultSet.next()) {
-                return null;
-            }
-            long rowId = resultSet.getLong(1);
-            if (resultSet.next() && resultSet.getLong(1) != rowId) {
-                throw new SQLException("ClickHouse database lock resolves to multiple rows");
-            }
-            return rowId;
-        }
-    }
-
-    private boolean hasCommittedReference(ReferenceKind kind, int id, String value) throws SQLException {
-        ClickHouseFamily family = referenceFamily(kind);
-        String column = referenceValueColumn(kind);
-        String sql = "SELECT id,`" + column + "` FROM " + tablePrefix + family.getTableName() + " WHERE id=? OR `" + column + "`=?";
-        boolean found = false;
-        try (Connection connection = openConnection(); PreparedStatement statement = connection.prepareStatement(sql)) {
-            statement.setInt(1, id);
-            statement.setString(2, value);
-            try (ResultSet resultSet = statement.executeQuery()) {
-                while (resultSet.next()) {
-                    int storedId = resultSet.getInt(1);
-                    String storedValue = resultSet.getString(2);
-                    if (storedId != id || !value.equals(storedValue)) {
-                        throw referenceConflict(kind, id, value, storedId, storedValue);
-                    }
-                    found = true;
-                }
-            }
-        }
-        return found;
-    }
-
-    private static ClickHouseFamily referenceFamily(ReferenceKind kind) {
+    static ClickHouseFamily referenceFamily(ReferenceKind kind) {
         switch (kind) {
             case ART:
                 return ClickHouseFamily.ART_MAP;
@@ -605,7 +583,7 @@ public final class ClickHouseConsumerWriteBatch implements ConsumerWriteBatch {
         }
     }
 
-    private static String referenceValueColumn(ReferenceKind kind) {
+    static String referenceValueColumn(ReferenceKind kind) {
         switch (kind) {
             case ART:
                 return "art";
@@ -662,10 +640,12 @@ public final class ClickHouseConsumerWriteBatch implements ConsumerWriteBatch {
         for (Map.Entry<ClickHouseFamily, Map<Long, ClickHouseEventPointer>> entry : localPointers.entrySet()) {
             pointers.put(entry.getKey(), new LinkedHashMap<>(entry.getValue()));
         }
-        return new AdapterCheckpoint(pointers, localBlockActions, usersByName, usersByUuid, userAliases, usernameHistory, localReferences, databaseLockRowId);
+        return new AdapterCheckpoint(pointers, localBlockActions, usersByName, usersByUuid, userAliases,
+                usernameHistory, localReferences, unpublishedReferences);
     }
 
     private void restore(AdapterCheckpoint checkpoint) {
+        discardReferenceCache(checkpoint.unpublishedReferences);
         localPointers.clear();
         for (Map.Entry<ClickHouseFamily, Map<Long, ClickHouseEventPointer>> entry : checkpoint.localPointers.entrySet()) {
             localPointers.put(entry.getKey(), new LinkedHashMap<>(entry.getValue()));
@@ -682,7 +662,19 @@ public final class ClickHouseConsumerWriteBatch implements ConsumerWriteBatch {
         usernameHistory.addAll(checkpoint.usernameHistory);
         localReferences.clear();
         checkpoint.localReferences.forEach((kind, references) -> localReferences.put(kind, new HashMap<>(references)));
-        databaseLockRowId = checkpoint.databaseLockRowId;
+        unpublishedReferences.clear();
+        checkpoint.unpublishedReferences.forEach((kind, references) -> unpublishedReferences.put(kind, new HashMap<>(references)));
+    }
+
+    private void discardReferenceCache(Map<ReferenceKind, Map<Integer, String>> retainedReferences) {
+        unpublishedReferences.forEach((kind, references) -> {
+            Map<Integer, String> retained = retainedReferences.get(kind);
+            references.forEach((id, value) -> {
+                if (retained == null || !value.equals(retained.get(id))) {
+                    ConfigHandler.uncacheIdentifier(kind, id, value);
+                }
+            });
+        });
     }
 
     private void clearBatchState() {
@@ -693,7 +685,7 @@ public final class ClickHouseConsumerWriteBatch implements ConsumerWriteBatch {
         userAliases.clear();
         usernameHistory.clear();
         localReferences.clear();
-        databaseLockRowId = null;
+        unpublishedReferences.clear();
     }
 
     private void ensureOpen() {
@@ -777,6 +769,17 @@ public final class ClickHouseConsumerWriteBatch implements ConsumerWriteBatch {
         }
     }
 
+    private static final class UserResolution {
+
+        private final UserRow row;
+        private final boolean changed;
+
+        private UserResolution(UserRow row, boolean changed) {
+            this.row = row;
+            this.changed = changed;
+        }
+    }
+
     private static final class AdapterCheckpoint {
 
         private final EnumMap<ClickHouseFamily, Map<Long, ClickHouseEventPointer>> localPointers;
@@ -786,9 +789,13 @@ public final class ClickHouseConsumerWriteBatch implements ConsumerWriteBatch {
         private final Map<String, Integer> userAliases;
         private final Set<String> usernameHistory;
         private final EnumMap<ReferenceKind, Map<Integer, String>> localReferences = new EnumMap<>(ReferenceKind.class);
-        private final Long databaseLockRowId;
+        private final EnumMap<ReferenceKind, Map<Integer, String>> unpublishedReferences = new EnumMap<>(ReferenceKind.class);
 
-        private AdapterCheckpoint(EnumMap<ClickHouseFamily, Map<Long, ClickHouseEventPointer>> localPointers, Map<Long, Integer> localBlockActions, Map<String, UserRow> usersByName, Map<String, UserRow> usersByUuid, Map<String, Integer> userAliases, Set<String> usernameHistory, EnumMap<ReferenceKind, Map<Integer, String>> localReferences, Long databaseLockRowId) {
+        private AdapterCheckpoint(EnumMap<ClickHouseFamily, Map<Long, ClickHouseEventPointer>> localPointers, Map<Long, Integer> localBlockActions,
+                Map<String, UserRow> usersByName, Map<String, UserRow> usersByUuid,
+                Map<String, Integer> userAliases, Set<String> usernameHistory,
+                EnumMap<ReferenceKind, Map<Integer, String>> localReferences,
+                EnumMap<ReferenceKind, Map<Integer, String>> unpublishedReferences) {
             this.localPointers = localPointers;
             this.localBlockActions = new HashMap<>(localBlockActions);
             this.usersByName = new HashMap<>(usersByName);
@@ -796,7 +803,7 @@ public final class ClickHouseConsumerWriteBatch implements ConsumerWriteBatch {
             this.userAliases = new HashMap<>(userAliases);
             this.usernameHistory = new HashSet<>(usernameHistory);
             localReferences.forEach((kind, references) -> this.localReferences.put(kind, new HashMap<>(references)));
-            this.databaseLockRowId = databaseLockRowId;
+            unpublishedReferences.forEach((kind, references) -> this.unpublishedReferences.put(kind, new HashMap<>(references)));
         }
     }
 

src/main/java/net/coreprotect/database/clickhouse/ClickHouseDatabase.java

@@ -8,11 +8,14 @@ import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.EnumMap;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
 import java.util.UUID;
 
+import net.coreprotect.config.Config;
 import net.coreprotect.config.ConfigHandler;
+import net.coreprotect.database.ConsumerWriteBatch.ReferenceKind;
 
 public final class ClickHouseDatabase implements AutoCloseable {
 
@@ -24,6 +27,7 @@ public final class ClickHouseDatabase implements AutoCloseable {
     private final String prefix;
     private final UUID datasetId;
     private final ClickHouseIdentityAllocator identityAllocator;
+    private final ClickHouseIdentityReservation identifierReservation;
     private final ClickHouseNativeClient nativeClient;
     private final ClickHouseBatchPublisher publisher;
     private final ClickHouseHighWaterPublisher highWaterPublisher;
@@ -33,13 +37,14 @@ public final class ClickHouseDatabase implements AutoCloseable {
     private boolean closed;
 
     private ClickHouseDatabase(ClickHouseJdbc jdbc, ClickHouseNativeClient nativeClient, String database, String prefix, UUID datasetId,
-            ClickHouseIdentityAllocator identityAllocator, ClickHouseWriterRegistration writerRegistration) {
+            ClickHouseIdentityAllocator identityAllocator, ClickHouseIdentityReservation identifierReservation, ClickHouseWriterRegistration writerRegistration) {
         this.jdbc = jdbc;
         this.nativeClient = nativeClient;
         this.database = database;
         this.prefix = prefix;
         this.datasetId = datasetId;
         this.identityAllocator = identityAllocator;
+        this.identifierReservation = identifierReservation;
         this.writerRegistration = writerRegistration;
         publisher = new ClickHouseBatchPublisher(jdbc, nativeClient, writerRegistration, database, prefix);
         highWaterPublisher = new ClickHouseHighWaterPublisher(jdbc, nativeClient, writerRegistration, database, prefix);
@@ -60,9 +65,13 @@ public final class ClickHouseDatabase implements AutoCloseable {
         ClickHouseNativeClient nativeClient = null;
         ClickHouseWriterRegistration writerRegistration = null;
         try {
+            writerRegistration = new ClickHouseWriterRegistration(controlDirectory);
+            writerRegistration.acquire();
+            UUID writerId = UUID.randomUUID();
             nativeClient = new ClickHouseNativeClient(config);
             List<String> statements = ClickHouseSchema.createStatements(config.getDatabase(), validatedPrefix);
             ClickHouseStorageIdentity storageIdentity;
+            ClickHouseHighWaterMarks remoteMarks;
             try (Connection connection = jdbc.openConnection()) {
                 requireServerVersion(connection);
                 requirePersistentDatabaseIdentity(connection, config.getDatabase());
@@ -70,27 +79,16 @@ public final class ClickHouseDatabase implements AutoCloseable {
                     jdbc.executeDdl(connection, statements.get(index));
                 }
                 ClickHouseSchema.validatePhysicalSchema(connection, config.getDatabase(), validatedPrefix);
-                storageIdentity = ClickHouseStorageIdentity.load(connection, config.getDatabase(), validatedPrefix);
-                if (storageIdentity == null) {
-                    ClickHouseSchema.requireUnownedTablesEmpty(connection, config.getDatabase(), validatedPrefix);
-                    storageIdentity = ClickHouseStorageIdentity.loadOrCreate(connection, config.getDatabase(), validatedPrefix);
-                }
-            }
-
-            writerRegistration = new ClickHouseWriterRegistration(jdbc, config.getDatabase(), validatedPrefix, storageIdentity.getDatasetId(), controlDirectory);
-            writerRegistration.acquire();
-            ClickHouseHighWaterMarks remoteMarks;
-            try (Connection connection = jdbc.openConnection()) {
+                storageIdentity = loadStorageIdentity(connection, config.getDatabase(), validatedPrefix);
                 for (int index = ClickHouseSchema.PHYSICAL_TABLE_COUNT; index < statements.size(); index++) {
                     jdbc.executeDdl(connection, statements.get(index));
                 }
                 remoteMarks = ClickHouseStartupReconciler.readRemote(connection, config.getDatabase(), validatedPrefix);
             }
-            ClickHouseIdentityAllocator identityAllocator = new ClickHouseIdentityAllocator(storageIdentity.getDatasetId(), remoteMarks);
-            ClickHouseDatabase database = new ClickHouseDatabase(jdbc, nativeClient, config.getDatabase(), validatedPrefix, storageIdentity.getDatasetId(), identityAllocator,
+            ClickHouseIdentityReservation reservation = new ClickHouseIdentityReservation(jdbc, config.getDatabase(), validatedPrefix, writerId);
+            ClickHouseIdentityAllocator identityAllocator = new ClickHouseIdentityAllocator(storageIdentity.getDatasetId(), reservation, remoteMarks);
+            return new ClickHouseDatabase(jdbc, nativeClient, config.getDatabase(), validatedPrefix, storageIdentity.getDatasetId(), identityAllocator, reservation,
                     writerRegistration);
-            database.retention.recoverAbandonedTargets();
-            return database;
         }
         catch (SQLException | RuntimeException exception) {
             try {
@@ -115,14 +113,36 @@ public final class ClickHouseDatabase implements AutoCloseable {
         }
     }
 
+    static ClickHouseStorageIdentity loadStorageIdentity(Connection connection, String database, String prefix) throws SQLException {
+        ClickHouseStorageIdentity storageIdentity = ClickHouseStorageIdentity.load(connection, database, prefix);
+        if (storageIdentity != null) {
+            return storageIdentity;
+        }
+        try {
+            ClickHouseSchema.requireUnownedTablesEmpty(connection, database, prefix);
+        }
+        catch (SQLException unownedTables) {
+            storageIdentity = ClickHouseStorageIdentity.load(connection, database, prefix);
+            if (storageIdentity == null) {
+                throw unownedTables;
+            }
+            return storageIdentity;
+        }
+        return ClickHouseStorageIdentity.loadOrCreate(connection, database, prefix);
+    }
+
     public Connection openConnection() throws SQLException {
         ensureOpen();
         return jdbc.openConnection();
     }
 
+    Connection openAuxiliaryConnection() throws SQLException {
+        ensureOpen();
+        return jdbc.openAuxiliaryConnection();
+    }
+
     public ClickHouseWriteBatch newWriteBatch() throws SQLException {
         ensureOpen();
-        writerRegistration.ensureOwned();
         return new ClickHouseWriteBatch(identityAllocator.nextBatchIdentity(), identityAllocator);
     }
 
@@ -131,6 +151,105 @@ public final class ClickHouseDatabase implements AutoCloseable {
         return publisher.publish(batch);
     }
 
+    public int nextIdentifierId(ReferenceKind kind, String value, int currentMaximum) throws SQLException {
+        ensureOpen();
+        Objects.requireNonNull(kind, "kind");
+        Objects.requireNonNull(value, "value");
+        if (currentMaximum < 0) {
+            throw new IllegalArgumentException("ClickHouse identifier maximums cannot be negative");
+        }
+        String family = ClickHouseConsumerWriteBatch.referenceFamily(kind).getTableName();
+        String identity = "canonical:" + family + ":" + value;
+        long id = identifierReservation.canonicalId(identity, () -> identifierReservation.next(family, currentMaximum));
+        return toIdentifier(id, family);
+    }
+
+    public String findIdentifierValue(ReferenceKind kind, int id) throws SQLException {
+        ensureOpen();
+        Objects.requireNonNull(kind, "kind");
+        if (id < 0) {
+            throw new IllegalArgumentException("ClickHouse identifiers cannot be negative");
+        }
+        ClickHouseFamily family = ClickHouseConsumerWriteBatch.referenceFamily(kind);
+        String column = ClickHouseConsumerWriteBatch.referenceValueColumn(kind);
+        String table = ClickHouseIdentifiers.qualified(database, prefix + family.getTableName());
+        String sql = "SELECT id,`" + column + "` FROM " + table + " WHERE id=? GROUP BY id,`" + column + "` LIMIT 2";
+        try (Connection connection = jdbc.openAuxiliaryConnection(); PreparedStatement statement = connection.prepareStatement(sql)) {
+            statement.setInt(1, id);
+            try (ResultSet resultSet = statement.executeQuery()) {
+                if (!resultSet.next()) {
+                    return null;
+                }
+                String value = resultSet.getString(2);
+                if (value == null) {
+                    throw new SQLException("ClickHouse " + family.getTableName() + " identifier " + id + " has no value");
+                }
+                if (resultSet.next()) {
+                    throw new SQLException("ClickHouse " + family.getTableName() + " identifier " + id + " resolves to multiple values");
+                }
+                return value;
+            }
+        }
+    }
+
+    public int findIdentifierId(ReferenceKind kind, String value) throws SQLException {
+        ensureOpen();
+        Objects.requireNonNull(kind, "kind");
+        Objects.requireNonNull(value, "value");
+        ClickHouseFamily family = ClickHouseConsumerWriteBatch.referenceFamily(kind);
+        String column = ClickHouseConsumerWriteBatch.referenceValueColumn(kind);
+        String table = ClickHouseIdentifiers.qualified(database, prefix + family.getTableName());
+        String sql = "SELECT id FROM " + table + " WHERE `" + column + "`=? AND id IS NOT NULL GROUP BY id LIMIT 2";
+        try (Connection connection = jdbc.openAuxiliaryConnection(); PreparedStatement statement = connection.prepareStatement(sql)) {
+            statement.setString(1, value);
+            try (ResultSet resultSet = statement.executeQuery()) {
+                if (!resultSet.next()) {
+                    return -1;
+                }
+                int id = toIdentifier(resultSet.getLong(1), family.getTableName());
+                if (resultSet.next()) {
+                    throw new SQLException("ClickHouse " + family.getTableName() + " value resolves to multiple identifiers: " + value);
+                }
+                return id;
+            }
+        }
+    }
+
+    int canonicalUserId(String user, String uuid, Integer existingId) throws SQLException {
+        ensureOpen();
+        String normalizedUser = Objects.requireNonNull(user, "user").toLowerCase(Locale.ROOT);
+        String normalizedUuid = Objects.requireNonNull(uuid, "uuid");
+        if (Config.getGlobal().DATABASE_LOCK) {
+            long id = existingId != null ? existingId : identityAllocator.nextRowId(ClickHouseFamily.USER);
+            return toIdentifier(id, ClickHouseFamily.USER.getTableName());
+        }
+        String nameIdentity = "canonical:user:name:" + normalizedUser;
+        if (normalizedUuid.isEmpty()) {
+            if (existingId != null) {
+                return existingId;
+            }
+            return toIdentifier(identifierReservation.canonicalId(nameIdentity,
+                    () -> identityAllocator.nextRowId(ClickHouseFamily.USER)),
+                    ClickHouseFamily.USER.getTableName());
+        }
+
+        String uuidIdentity = "canonical:user:uuid:" + normalizedUuid;
+        long uuidId = identifierReservation.canonicalId(uuidIdentity,
+                () -> existingId != null
+                        ? existingId
+                        : identifierReservation.canonicalId(nameIdentity,
+                                () -> identityAllocator.nextRowId(ClickHouseFamily.USER)));
+        identifierReservation.canonicalId(nameIdentity, uuidId);
+        return toIdentifier(uuidId, ClickHouseFamily.USER.getTableName());
+    }
+
+    private static int toIdentifier(long id, String family) throws SQLException {
+        if (id < 1 || id > Integer.MAX_VALUE) {
+            throw new SQLException("ClickHouse " + family + " identifier is outside CoreProtect's positive signed 32-bit range: " + id);
+        }
+        return (int) id;
+    }
+
     public void preserveCompatibilityHighWaterMarks(Map<ClickHouseFamily, Long> highWaterMarks) throws SQLException {
         ensureOpen();
         Objects.requireNonNull(highWaterMarks, "highWaterMarks");

src/main/java/net/coreprotect/database/clickhouse/ClickHouseEventBatch.java

@@ -6,6 +6,7 @@ import java.util.EnumMap;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Objects;
+import java.util.TreeMap;
 import java.util.UUID;
 
 import net.coreprotect.database.DatabaseType;
@@ -15,10 +16,13 @@ import net.coreprotect.utility.serialize.EntityDataCodec;
 
 public final class ClickHouseEventBatch implements AutoCloseable {
 
+    private static final long SINGLETON_ROW_ID = 1L;
+
     private final ClickHouseBatchIdentity identity;
     private final ClickHouseRowIdAllocator rowIdAllocator;
     private final ClickHouseRowBinaryBuffer rows = new ClickHouseRowBinaryBuffer(ClickHouseSchema.EVENT_COLUMNS, ClickHouseSchema.EVENT_COLUMN_TYPES);
     private final EnumMap<ClickHouseFamily, HashSet<Long>> versionRowIds = new EnumMap<>(ClickHouseFamily.class);
+    private final TreeMap<Integer, Integer> partitionRowCounts = new TreeMap<>();
     private int eventCount;
     private int duplicateVersionCount;
     private int currentTime;
@@ -92,15 +96,11 @@ public final class ClickHouseEventBatch implements AutoCloseable {
     }
 
     public long addDatabaseLock(int time, int status) throws SQLException {
-        long rowId = beginRow(ClickHouseFamily.DATABASE_LOCK, time);
-        set("status", status);
-        set("database_lock_time", time);
-        commitRow(ClickHouseFamily.DATABASE_LOCK, rowId);
-        return rowId;
+        return addDatabaseLockVersion(SINGLETON_ROW_ID, time, status);
     }
 
     public long addDatabaseLockVersion(long rowId, int time, int status) throws SQLException {
-        beginRow(ClickHouseFamily.DATABASE_LOCK, rowId, time);
+        beginExplicitRow(ClickHouseFamily.DATABASE_LOCK, rowId, time);
         set("status", status);
         set("database_lock_time", time);
         commitRow(ClickHouseFamily.DATABASE_LOCK, rowId);
@@ -149,7 +149,7 @@ public final class ClickHouseEventBatch implements AutoCloseable {
     }
 
     public long addBlockDataMap(int id, String blockData) throws SQLException {
-        long rowId = beginRow(ClickHouseFamily.BLOCKDATA_MAP, 0);
+        long rowId = beginExplicitRow(ClickHouseFamily.BLOCKDATA_MAP, id, 0);
         set("id", id);
         set("text", Objects.requireNonNull(blockData, "blockData"));
         commitRow(ClickHouseFamily.BLOCKDATA_MAP, rowId);
@@ -191,15 +191,8 @@ public final class ClickHouseEventBatch implements AutoCloseable {
         return rowId;
     }
 
-    public long addUser(int time, String userName, String uuid) throws SQLException {
-        long rowId = beginRow(ClickHouseFamily.USER, time);
-        setUser(userName, uuid);
-        commitRow(ClickHouseFamily.USER, rowId);
-        return rowId;
-    }
-
     public long addUserVersion(long rowId, int time, String userName, String uuid) throws SQLException {
-        beginRow(ClickHouseFamily.USER, rowId, time);
+        beginExplicitRow(ClickHouseFamily.USER, rowId, time);
         setUser(userName, uuid);
         commitRow(ClickHouseFamily.USER, rowId);
         return rowId;
@@ -213,14 +206,11 @@ public final class ClickHouseEventBatch implements AutoCloseable {
     }
 
     public long addVersion(int time, String version) throws SQLException {
-        long rowId = beginRow(ClickHouseFamily.VERSION, time);
-        set("version", Objects.requireNonNull(version, "version"));
-        commitRow(ClickHouseFamily.VERSION, rowId);
-        return rowId;
+        return addVersionRevision(SINGLETON_ROW_ID, time, version);
     }
 
     public long addVersionRevision(long rowId, int time, String version) throws SQLException {
-        beginRow(ClickHouseFamily.VERSION, rowId, time);
+        beginExplicitRow(ClickHouseFamily.VERSION, rowId, time);
         set("version", Objects.requireNonNull(version, "version"));
         commitRow(ClickHouseFamily.VERSION, rowId);
         return rowId;
@@ -237,7 +227,7 @@ public final class ClickHouseEventBatch implements AutoCloseable {
             throw new IllegalArgumentException("ClickHouse core data requires a dedicated writer: " + family.getTableName());
         }
         int time = numberOrZero(values.get("time")).intValue();
-        beginRow(family, rowId, time);
+        beginExplicitRow(family, rowId, time);
         if (family == ClickHouseFamily.ENTITY_SPAWN) {
             set("x", originKey(values.get("origin_x")));
             set("z", originKey(values.get("origin_z")));
@@ -282,14 +272,17 @@ public final class ClickHouseEventBatch implements AutoCloseable {
         return eventCount - duplicateVersionCount;
     }
 
-    void seal(ClickHouseStateBatch state) throws SQLException {
+    Map<Integer, Integer> seal(ClickHouseStateBatch state) throws SQLException {
         if (sealed) {
-            return;
+            return partitionRowCounts;
         }
         ensureWritable();
-        Objects.requireNonNull(state, "state").appendTo(rows, eventCount);
+        ClickHouseStateBatch requiredState = Objects.requireNonNull(state, "state");
+        requiredState.appendTo(rows, eventCount, partitionRowCounts);
+        appendBatchReceipts(eventCount + requiredState.getRollbackCount() + requiredState.getEntityStateCount());
         rows.seal();
         sealed = true;
+        return partitionRowCounts;
     }
 
     InputStream openRows() {
@@ -299,6 +292,13 @@ public final class ClickHouseEventBatch implements AutoCloseable {
         return rows.openStream();
     }
 
+    InputStream openRows(int partitionId) {
+        if (!sealed) {
+            throw new IllegalStateException("ClickHouse event batch is not sealed");
+        }
+        return rows.openStream(partitionId);
+    }
+
     ClickHouseEventPointer getLastPointer() {
         if (lastPointer == null) {
             throw new IllegalStateException("ClickHouse event batch has no rows");
@@ -308,7 +308,7 @@ public final class ClickHouseEventBatch implements AutoCloseable {
 
     Checkpoint checkpoint() {
         ensureWritable();
-        return new Checkpoint(rows.checkpoint(), eventCount, duplicateVersionCount, versionRowIds, lastPointer);
+        return new Checkpoint(rows.checkpoint(), eventCount, duplicateVersionCount, versionRowIds, partitionRowCounts, lastPointer);
     }
 
     void restore(Checkpoint checkpoint) {
@@ -319,6 +319,8 @@ public final class ClickHouseEventBatch implements AutoCloseable {
         duplicateVersionCount = checkpoint.duplicateVersionCount;
         versionRowIds.clear();
         checkpoint.versionRowIds.forEach((family, rowIds) -> versionRowIds.put(family, new HashSet<>(rowIds)));
+        partitionRowCounts.clear();
+        partitionRowCounts.putAll(checkpoint.partitionRowCounts);
         lastPointer = checkpoint.lastPointer;
     }
 
@@ -330,7 +332,7 @@ public final class ClickHouseEventBatch implements AutoCloseable {
     }
 
     private long addNamedMap(ClickHouseFamily family, int id, String name) throws SQLException {
-        long rowId = beginRow(family, 0);
+        long rowId = beginExplicitRow(family, id, 0);
         set("id", id);
         set("name", Objects.requireNonNull(name, family.getTableName()));
         commitRow(family, rowId);
@@ -359,10 +361,16 @@ public final class ClickHouseEventBatch implements AutoCloseable {
         return rowId;
     }
 
-    private long beginRow(ClickHouseFamily family, int time) {
+    private long beginRow(ClickHouseFamily family, int time) throws SQLException {
         return beginRow(family, rowIdAllocator.nextRowId(family), time);
     }
 
+    private long beginExplicitRow(ClickHouseFamily family, long rowId, int time) {
+        ensureWritable();
+        rowIdAllocator.observeRowId(family, rowId);
+        return beginRow(family, rowId, time);
+    }
+
     private long beginRow(ClickHouseFamily family, long rowId, int time) {
         ensureWritable();
         Objects.requireNonNull(family, "family");
@@ -372,7 +380,6 @@ public final class ClickHouseEventBatch implements AutoCloseable {
         if (rowId < 1) {
             throw new IllegalArgumentException("ClickHouse compatibility row IDs must be positive");
         }
-        rowIdAllocator.observeRowId(family, rowId);
         rows.beginRow();
         currentTime = time;
         currentWorldId = 0;
@@ -397,7 +404,7 @@ public final class ClickHouseEventBatch implements AutoCloseable {
 
     private void setUser(String userName, String uuid) {
         set("user_name", Objects.requireNonNull(userName, "userName"));
-        set("uuid", uuid == null ? "" : uuid);
+        set("uuid", uuid == null || uuid.isEmpty() ? null : uuid);
     }
 
     private void setBinary(String column, byte[] value) {
@@ -476,7 +483,9 @@ public final class ClickHouseEventBatch implements AutoCloseable {
     }
 
     private void commitRow(ClickHouseFamily family, long rowId) throws SQLException {
-        rows.commitRow(family.getTableName() + " event");
+        int partitionId = ClickHouseSchema.eventPartitionId(family, currentTime);
+        rows.commitRow(family.getTableName() + " event", partitionId);
+        partitionRowCounts.merge(partitionId, 1, Math::addExact);
         if (isVersionedFamily(family) && !versionRowIds.computeIfAbsent(family, ignored -> new HashSet<>()).add(rowId)) {
             duplicateVersionCount++;
         }
@@ -484,6 +493,21 @@ public final class ClickHouseEventBatch implements AutoCloseable {
         eventCount++;
     }
 
+    private void appendBatchReceipts(int firstOrdinal) throws SQLException {
+        int ordinal = firstOrdinal;
+        for (Map.Entry<Integer, Integer> partition : partitionRowCounts.entrySet()) {
+            rows.beginRow();
+            rows.set("batch_sequence", identity.getBatchSequence());
+            rows.set("batch_id", identity.getBatchId());
+            rows.set("batch_ordinal", ordinal++);
+            rows.set("family", ClickHouseSchema.BATCH_RECEIPT_FAMILY);
+            rows.set("rowid", identity.getBatchSequence());
+            rows.set("wid", partition.getKey());
+            rows.set("amount", partition.getValue());
+            rows.commitRow("batch receipt", partition.getKey());
+        }
+    }
+
     private static boolean isVersionedFamily(ClickHouseFamily family) {
         return family == ClickHouseFamily.DATABASE_LOCK || family == ClickHouseFamily.USER || family == ClickHouseFamily.VERSION;
     }
@@ -558,13 +582,16 @@ public final class ClickHouseEventBatch implements AutoCloseable {
         private final int eventCount;
         private final int duplicateVersionCount;
         private final EnumMap<ClickHouseFamily, HashSet<Long>> versionRowIds = new EnumMap<>(ClickHouseFamily.class);
+        private final TreeMap<Integer, Integer> partitionRowCounts;
         private final ClickHouseEventPointer lastPointer;
 
-        private Checkpoint(int bufferSize, int eventCount, int duplicateVersionCount, EnumMap<ClickHouseFamily, HashSet<Long>> versionRowIds, ClickHouseEventPointer lastPointer) {
+        private Checkpoint(int bufferSize, int eventCount, int duplicateVersionCount, EnumMap<ClickHouseFamily, HashSet<Long>> versionRowIds,
+                Map<Integer, Integer> partitionRowCounts, ClickHouseEventPointer lastPointer) {
             this.bufferSize = bufferSize;
             this.eventCount = eventCount;
             this.duplicateVersionCount = duplicateVersionCount;
             versionRowIds.forEach((family, rowIds) -> this.versionRowIds.put(family, new HashSet<>(rowIds)));
+            this.partitionRowCounts = new TreeMap<>(partitionRowCounts);
             this.lastPointer = lastPointer;
         }
     }

src/main/java/net/coreprotect/database/clickhouse/ClickHouseIdentityAllocator.java

@@ -1,34 +1,38 @@
 package net.coreprotect.database.clickhouse;
 
-import java.util.EnumMap;
+import java.sql.SQLException;
 import java.util.Objects;
 import java.util.UUID;
-import java.util.concurrent.atomic.AtomicLong;
 
 public final class ClickHouseIdentityAllocator implements ClickHouseRowIdAllocator {
 
     private final UUID datasetId;
-    private final AtomicLong batchSequence;
-    private final EnumMap<ClickHouseFamily, AtomicLong> rowIds = new EnumMap<>(ClickHouseFamily.class);
+    private final ClickHouseIdentityReservation reservation;
 
-    public ClickHouseIdentityAllocator(UUID datasetId, ClickHouseHighWaterMarks highWaterMarks) {
+    ClickHouseIdentityAllocator(UUID datasetId, ClickHouseIdentityReservation reservation, ClickHouseHighWaterMarks highWaterMarks) {
         this.datasetId = Objects.requireNonNull(datasetId, "datasetId");
+        this.reservation = Objects.requireNonNull(reservation, "reservation");
         Objects.requireNonNull(highWaterMarks, "highWaterMarks");
-        batchSequence = new AtomicLong(highWaterMarks.getBatchSequence());
+        if (highWaterMarks.getBatchSequence() > 0) {
+            reservation.observe(ClickHouseIdentityReservation.BATCH_SEQUENCE, highWaterMarks.getBatchSequence());
+        }
         for (ClickHouseFamily family : ClickHouseFamily.values()) {
-            rowIds.put(family, new AtomicLong(highWaterMarks.getCompatibilityRowId(family)));
+            long rowId = highWaterMarks.getCompatibilityRowId(family);
+            if (rowId > 0) {
+                reservation.observe(family.getTableName(), rowId);
+            }
         }
     }
 
-    public ClickHouseBatchIdentity nextBatchIdentity() {
-        long sequence = increment(batchSequence, "batch sequence");
+    public ClickHouseBatchIdentity nextBatchIdentity() throws SQLException {
+        long sequence = reservation.next(ClickHouseIdentityReservation.BATCH_SEQUENCE);
         return ClickHouseBatchIdentity.create(datasetId, sequence);
     }
 
     @Override
-    public long nextRowId(ClickHouseFamily family) {
+    public long nextRowId(ClickHouseFamily family) throws SQLException {
         Objects.requireNonNull(family, "family");
-        long rowId = increment(rowIds.get(family), family.getTableName() + " row ID");
+        long rowId = reservation.next(family.getTableName());
         validateRowId(family, rowId);
         return rowId;
     }
@@ -37,7 +41,7 @@ public final class ClickHouseIdentityAllocator implements ClickHouseRowIdAllocat
     public void observeRowId(ClickHouseFamily family, long rowId) {
         Objects.requireNonNull(family, "family");
         validateRowId(family, rowId);
-        rowIds.get(family).accumulateAndGet(rowId, Math::max);
+        reservation.observe(family.getTableName(), rowId);
     }
 
     private static void validateRowId(ClickHouseFamily family, long rowId) {
@@ -49,12 +53,4 @@ public final class ClickHouseIdentityAllocator implements ClickHouseRowIdAllocat
         }
     }
 
-    private static long increment(AtomicLong value, String name) {
-        long next = value.incrementAndGet();
-        if (next < 1) {
-            throw new IllegalStateException("ClickHouse " + name + " exhausted its signed 64-bit range");
-        }
-        return next;
-    }
-
 }

src/main/java/net/coreprotect/database/clickhouse/ClickHouseIdentityReservation.java

@@ -0,0 +1,258 @@
+package net.coreprotect.database.clickhouse;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Map;
+import java.util.Objects;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+
+final class ClickHouseIdentityReservation {
+
+    static final String BATCH_SEQUENCE = "batch_sequence";
+
+    private static final String CLAIM_ORDER = "_block_number,block_start,toString(writer_id)";
+
+    private static final long BLOCK_SIZE = 4_096L;
+    private static final long RESERVATION_TIMEOUT_NANOS = TimeUnit.SECONDS.toNanos(30);
+
+    private final ClickHouseJdbc jdbc;
+    private final String table;
+    private final UUID writerId;
+    private final Map<String, Sequence> sequences = new ConcurrentHashMap<>();
+
+    ClickHouseIdentityReservation(ClickHouseJdbc jdbc, String database, String prefix, UUID writerId) {
+        this.jdbc = Objects.requireNonNull(jdbc, "jdbc");
+        this.writerId = Objects.requireNonNull(writerId, "writerId");
+        table = ClickHouseIdentifiers.qualified(database, prefix + "identity_reservation");
+    }
+
+    long next(String sequence) throws SQLException {
+        return next(sequence, 0L);
+    }
+
+    long next(String sequence, long floor) throws SQLException {
+        Objects.requireNonNull(sequence, "sequence");
+        if (floor < 0) {
+            throw new IllegalArgumentException("ClickHouse identity floors cannot be negative");
+        }
+        return sequences.computeIfAbsent(sequence, Sequence::new).next(floor);
+    }
+
+    void observe(String sequence, long id) {
+        Objects.requireNonNull(sequence, "sequence");
+        if (id < 0) {
+            throw new IllegalArgumentException("ClickHouse identities cannot be negative");
+        }
+        sequences.computeIfAbsent(sequence, Sequence::new).observe(id);
+    }
+
+    long canonicalId(String identity, CandidateAllocator allocator) throws SQLException {
+        Objects.requireNonNull(allocator, "allocator");
+        Long claimed = claimedId(identity);
+        if (claimed != null) {
+            return claimed;
+        }
+        return claimCanonicalId(identity, allocator.next());
+    }
+
+    long canonicalId(String identity, long candidate) throws SQLException {
+        Long claimed = claimedId(identity);
+        if (claimed != null) {
+            return claimed;
+        }
+        return claimCanonicalId(identity, candidate);
+    }
+
+    private long claimCanonicalId(String identity, long candidate) throws SQLException {
+        if (candidate < 1) {
+            throw new SQLException("ClickHouse canonical identity allocator returned an invalid ID: " + candidate);
+        }
+        long deadline = reservationDeadline();
+        claimCanonical(identity, candidate, deadline);
+        Long claimed = claimedId(identity, deadline);
+        if (claimed == null) {
+            throw new SQLException("ClickHouse canonical identity is not visible after being claimed");
+        }
+        return claimed;
+    }
+
+    Long claimedId(String identity) throws SQLException {
+        return claimedId(identity, reservationDeadline());
+    }
+
+    private Long claimedId(String identity, long deadline) throws SQLException {
+        if (identity == null || identity.isEmpty()) {
+            throw new IllegalArgumentException("ClickHouse canonical identities cannot be empty");
+        }
+        String sql = "SELECT block_start FROM " + table + " WHERE sequence=? ORDER BY " + CLAIM_ORDER + " LIMIT 1";
+        try (Connection connection = jdbc.openAuxiliaryConnection(); PreparedStatement statement = connection.prepareStatement(sql)) {
+            setQueryTimeout(statement, deadline, identity);
+            statement.setString(1, identity);
+            try (ResultSet resultSet = statement.executeQuery()) {
+                return resultSet.next() ? resultSet.getLong(1) : null;
+            }
+        }
+    }
+
+    private Long readTopBlockEnd(String sequence, long deadline) throws SQLException {
+        String sql = "SELECT block_start FROM " + table + " WHERE sequence=? ORDER BY block_start DESC LIMIT 1";
+        try (Connection connection = jdbc.openAuxiliaryConnection(); PreparedStatement statement = connection.prepareStatement(sql)) {
+            setQueryTimeout(statement, deadline, sequence);
+            statement.setString(1, sequence);
+            try (ResultSet resultSet = statement.executeQuery()) {
+                if (!resultSet.next()) {
+                    return null;
+                }
+                long blockStart = resultSet.getLong(1);
+                if (blockStart < 0 || blockStart > Long.MAX_VALUE - BLOCK_SIZE) {
+                    throw new SQLException("ClickHouse identity reservation for " + sequence + " exceeds the supported range");
+                }
+                return blockStart + BLOCK_SIZE;
+            }
+        }
+    }
+
+    private UUID readBlockOwner(String sequence, long blockStart, long deadline) throws SQLException {
+        String sql = "SELECT " + winner("toString(writer_id)") + " FROM " + table + " WHERE sequence=? AND block_start=? GROUP BY block_start";
+        try (Connection connection = jdbc.openAuxiliaryConnection(); PreparedStatement statement = connection.prepareStatement(sql)) {
+            setQueryTimeout(statement, deadline, sequence);
+            statement.setString(1, sequence);
+            statement.setLong(2, blockStart);
+            try (ResultSet resultSet = statement.executeQuery()) {
+                if (!resultSet.next()) {
+                    throw new SQLException("ClickHouse identity block " + sequence + "@" + blockStart + " is not visible after being claimed");
+                }
+                return parseWriterId(sequence, resultSet.getString(1));
+            }
+        }
+    }
+
+    private void claimCanonical(String sequence, long blockStart, long deadline) throws SQLException {
+        String sql = "INSERT INTO " + table + " (sequence,block_start,writer_id) VALUES (?,?,?)";
+        try (Connection connection = jdbc.openAuxiliaryConnection(); PreparedStatement statement = connection.prepareStatement(sql)) {
+            setQueryTimeout(statement, deadline, sequence);
+            statement.setString(1, sequence);
+            statement.setLong(2, blockStart);
+            statement.setObject(3, writerId);
+            statement.execute();
+        }
+    }
+
+    private static void setQueryTimeout(PreparedStatement statement, long deadline, String sequence) throws SQLException {
+        long remaining = deadline - System.nanoTime();
+        if (remaining <= 0) {
+            throw new SQLException("Timed out reserving ClickHouse identity block " + sequence);
+        }
+        long seconds = Math.max(1L, TimeUnit.NANOSECONDS.toSeconds(remaining));
+        statement.setQueryTimeout((int) Math.min(seconds, Integer.MAX_VALUE));
+    }
+
+    private static long reservationDeadline() {
+        return System.nanoTime() + RESERVATION_TIMEOUT_NANOS;
+    }
+
+    private static String winner(String column) {
+        return "argMin(" + column + ",(" + CLAIM_ORDER + "))";
+    }
+
+    private static UUID parseWriterId(String sequence, String value) throws SQLException {
+        try {
+            return UUID.fromString(value);
+        }
+        catch (IllegalArgumentException | NullPointerException exception) {
+            throw new SQLException("ClickHouse identity reservation for " + sequence + " has an invalid writer ID", exception);
+        }
+    }
+
+    private final class Sequence {
+
+        private final String name;
+        private long next;
+        private long end;
+
+        private Sequence(String name) {
+            this.name = name;
+        }
+
+        private synchronized long next(long floor) throws SQLException {
+            observe(floor);
+            if (next >= end) {
+                reserve();
+            }
+            return next++;
+        }
+
+        private synchronized void observe(long id) {
+            if (id >= next) {
+                if (id == Long.MAX_VALUE) {
+                    throw new IllegalStateException("ClickHouse " + name + " identity range is exhausted");
+                }
+                next = id + 1;
+            }
+        }
+
+        private void reserve() throws SQLException {
+            long deadline = reservationDeadline();
+            int contentions = 0;
+            Long topEnd = readTopBlockEnd(name, deadline);
+            while (true) {
+                requireReservationActive(deadline);
+                long blockStart = Math.max(topEnd == null ? 0L : topEnd,
+                        containingBlockStart(next - 1));
+                if (blockStart > Long.MAX_VALUE - BLOCK_SIZE - 1) {
+                    throw new SQLException("ClickHouse " + name + " identity range is exhausted");
+                }
+                claimCanonical(name, blockStart, deadline);
+                boolean owned = writerId.equals(readBlockOwner(name, blockStart, deadline));
+                requireReservationActive(deadline);
+                if (owned) {
+                    next = Math.max(next, blockStart + 1);
+                    end = blockStart + BLOCK_SIZE + 1;
+                    return;
+                }
+                waitBeforeRetry(deadline, contentions++);
+                topEnd = readTopBlockEnd(name, deadline);
+            }
+        }
+
+        private void waitBeforeRetry(long deadline, int contentions) throws SQLException {
+            requireReservationActive(deadline);
+            long remaining = deadline - System.nanoTime();
+            long maximumDelay = 1L << Math.min(contentions, 3);
+            long delay = Math.min(remaining, TimeUnit.MILLISECONDS.toNanos(ThreadLocalRandom.current().nextLong(1, maximumDelay + 1)));
+            try {
+                TimeUnit.NANOSECONDS.sleep(delay);
+            }
+            catch (InterruptedException exception) {
+                Thread.currentThread().interrupt();
+                throw new SQLException("Interrupted while reserving ClickHouse identity block " + name, exception);
+            }
+            requireReservationActive(deadline);
+        }
+
+        private void requireReservationActive(long deadline) throws SQLException {
+            if (Thread.currentThread().isInterrupted()) {
+                throw new SQLException("Interrupted while reserving ClickHouse identity block " + name);
+            }
+            if (deadline - System.nanoTime() <= 0) {
+                throw new SQLException("Timed out reserving ClickHouse identity block " + name);
+            }
+        }
+
+        private long containingBlockStart(long id) {
+            return id - id % BLOCK_SIZE;
+        }
+    }
+
+    @FunctionalInterface
+    interface CandidateAllocator {
+
+        long next() throws SQLException;
+    }
+
+}

src/main/java/net/coreprotect/database/clickhouse/ClickHouseJdbc.java

@@ -16,24 +16,34 @@ public final class ClickHouseJdbc implements AutoCloseable {
     private static final Set<String> DDL_OPERATIONS = Set.of("ALTER", "CREATE", "DROP", "RENAME", "TRUNCATE");
 
     private final HikariDataSource dataSource;
+    private final HikariDataSource auxiliaryDataSource;
 
     public ClickHouseJdbc(ClickHouseJdbcConfig config) {
         Objects.requireNonNull(config, "config");
+        dataSource = pool(config, "CoreProtect-ClickHouse", 10);
+        auxiliaryDataSource = pool(config, "CoreProtect-ClickHouse-Identity", 4);
+    }
+
+    private static HikariDataSource pool(ClickHouseJdbcConfig config, String name, int maximumPoolSize) {
         HikariConfig pool = new HikariConfig();
         pool.setDataSource(new HikariClickHouseDataSource(config));
-        pool.setPoolName("CoreProtect-ClickHouse");
+        pool.setPoolName(name);
         pool.setMinimumIdle(0);
-        pool.setMaximumPoolSize(10);
+        pool.setMaximumPoolSize(maximumPoolSize);
         pool.setConnectionTimeout(10_000L);
         pool.setIdleTimeout(60_000L);
         pool.setInitializationFailTimeout(-1L);
-        dataSource = new HikariDataSource(pool);
+        return new HikariDataSource(pool);
     }
 
     public Connection openConnection() throws SQLException {
         return dataSource.getConnection();
     }
 
+    Connection openAuxiliaryConnection() throws SQLException {
+        return auxiliaryDataSource.getConnection();
+    }
+
     public void executeDdl(String ddl) throws SQLException {
         try (Connection connection = openConnection()) {
             executeDdl(connection, ddl);
@@ -50,7 +60,12 @@ public final class ClickHouseJdbc implements AutoCloseable {
 
     @Override
     public void close() {
-        dataSource.close();
+        try {
+            auxiliaryDataSource.close();
+        }
+        finally {
+            dataSource.close();
+        }
     }
 
     private static String requireSingleDdl(String ddl) {

src/main/java/net/coreprotect/database/clickhouse/ClickHouseRetention.java

@@ -40,12 +40,6 @@ final class ClickHouseRetention {
         highWaterTable = table("retention_high_water");
     }
 
-    synchronized void recoverAbandonedTargets() throws SQLException {
-        try (Connection connection = jdbc.openConnection()) {
-            cleanupAbandonedTargets(connection);
-        }
-    }
-
     void cancelPurge() {
         purgeCancellationRequested = true;
         Connection connection = activePurgeConnection;
@@ -177,7 +171,7 @@ final class ClickHouseRetention {
         String sql = "INSERT INTO " + highWaterTable
                 + " (batch_sequence,family,rowid,recorded_at)"
                 + " SELECT max(batch_sequence),family,max(rowid),now64(3, 'UTC')"
-                + " FROM " + eventTable + " GROUP BY family";
+                + " FROM " + eventTable + " WHERE family!='" + ClickHouseSchema.BATCH_RECEIPT_FAMILY + "' GROUP BY family";
         try (PreparedStatement statement = connection.prepareStatement(sql)) {
             statement.execute();
         }

src/main/java/net/coreprotect/database/clickhouse/ClickHouseRowBinaryBuffer.java

@@ -6,6 +6,7 @@ import java.io.IOException;
 import java.io.InputStream;
 import java.sql.SQLException;
 import java.time.LocalDateTime;
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -23,6 +24,7 @@ final class ClickHouseRowBinaryBuffer implements AutoCloseable {
     private final String[] types;
     private final Object[] defaults;
     private final Object[] values;
+    private final List<RowSpan> rowSpans = new ArrayList<>();
     private boolean sealed;
     private boolean failed;
     private boolean rowStarted;
@@ -63,12 +65,21 @@ final class ClickHouseRowBinaryBuffer implements AutoCloseable {
     }
 
     void commitRow(String description) throws SQLException {
+        commitRow(description, 0);
+    }
+
+    void commitRow(String description, int partitionId) throws SQLException {
         ensureWritable();
         requireActiveRow();
+        if (partitionId < 0) {
+            throw new IllegalArgumentException("ClickHouse row partition IDs cannot be negative");
+        }
+        int start = rows.size();
         try {
             for (int index = 0; index < values.length; index++) {
                 writeValue(types[index], values[index]);
             }
+            rowSpans.add(new RowSpan(partitionId, start, rows.size() - start));
             rowStarted = false;
         }
         catch (IOException | RuntimeException exception) {
@@ -92,6 +103,16 @@ final class ClickHouseRowBinaryBuffer implements AutoCloseable {
         return rows.openStream();
     }
 
+    InputStream openStream(int partitionId) {
+        if (!sealed) {
+            throw new IllegalStateException("ClickHouse RowBinary buffer is not sealed");
+        }
+        if (partitionId < 0) {
+            throw new IllegalArgumentException("ClickHouse row partition IDs cannot be negative");
+        }
+        return rows.openStream(rowSpans, partitionId);
+    }
+
     int checkpoint() {
         ensureWritable();
         if (rowStarted) {
@@ -108,6 +129,13 @@ final class ClickHouseRowBinaryBuffer implements AutoCloseable {
             throw new IllegalArgumentException("Invalid ClickHouse RowBinary checkpoint");
         }
         rows.truncate(checkpoint);
+        while (!rowSpans.isEmpty()) {
+            RowSpan span = rowSpans.get(rowSpans.size() - 1);
+            if (span.start + span.length <= checkpoint) {
+                break;
+            }
+            rowSpans.remove(rowSpans.size() - 1);
+        }
         failed = false;
         rowStarted = false;
     }
@@ -116,6 +144,7 @@ final class ClickHouseRowBinaryBuffer implements AutoCloseable {
     public void close() {
         sealed = true;
         rows.reset();
+        rowSpans.clear();
     }
 
     private void ensureWritable() {
@@ -253,9 +282,91 @@ final class ClickHouseRowBinaryBuffer implements AutoCloseable {
             return new ByteArrayInputStream(buf, 0, count);
         }
 
+        private InputStream openStream(List<RowSpan> spans, int partitionId) {
+            return new PartitionInputStream(buf, spans, partitionId);
+        }
+
         private void truncate(int size) {
             count = size;
         }
     }
 
+    private static final class RowSpan {
+
+        private final int partitionId;
+        private final int start;
+        private final int length;
+
+        private RowSpan(int partitionId, int start, int length) {
+            this.partitionId = partitionId;
+            this.start = start;
+            this.length = length;
+        }
+    }
+
+    private static final class PartitionInputStream extends InputStream {
+
+        private final byte[] data;
+        private final List<RowSpan> spans;
+        private final int partitionId;
+        private int spanIndex;
+        private int spanOffset;
+
+        private PartitionInputStream(byte[] data, List<RowSpan> spans, int partitionId) {
+            this.data = data;
+            this.spans = spans;
+            this.partitionId = partitionId;
+        }
+
+        @Override
+        public int read() {
+            RowSpan span = currentSpan();
+            if (span == null) {
+                return -1;
+            }
+            return data[span.start + spanOffset++] & 0xff;
+        }
+
+        @Override
+        public int read(byte[] output, int offset, int length) {
+            if (output == null) {
+                throw new NullPointerException("output");
+            }
+            if (offset < 0 || length < 0 || length > output.length - offset) {
+                throw new IndexOutOfBoundsException();
+            }
+            if (length == 0) {
+                return 0;
+            }
+
+            int copied = 0;
+            while (length > 0) {
+                RowSpan span = currentSpan();
+                if (span == null) {
+                    return copied == 0 ? -1 : copied;
+                }
+                int count = Math.min(length, span.length - spanOffset);
+                System.arraycopy(data, span.start + spanOffset, output, offset, count);
+                spanOffset += count;
+                offset += count;
+                length -= count;
+                copied += count;
+            }
+            return copied;
+        }
+
+        private RowSpan currentSpan() {
+            while (spanIndex < spans.size()) {
+                RowSpan span = spans.get(spanIndex);
+                if (span.partitionId != partitionId || spanOffset >= span.length) {
+                    spanIndex++;
+                    spanOffset = 0;
+                    continue;
+                }
+                return span;
+            }
+            return null;
+        }
+    }
+
 }

src/main/java/net/coreprotect/database/clickhouse/ClickHouseRowIdAllocator.java

@@ -1,9 +1,11 @@
 package net.coreprotect.database.clickhouse;
 
+import java.sql.SQLException;
+
 @FunctionalInterface
 public interface ClickHouseRowIdAllocator {
 
-    long nextRowId(ClickHouseFamily family);
+    long nextRowId(ClickHouseFamily family) throws SQLException;
 
     default void observeRowId(ClickHouseFamily family, long rowId) {
     }

src/main/java/net/coreprotect/database/clickhouse/ClickHouseSchema.java

@@ -4,6 +4,8 @@ import java.sql.Connection;
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -12,14 +14,17 @@ import java.util.StringJoiner;
 
 public final class ClickHouseSchema {
 
-    static final int VERSION = 3;
+    static final int VERSION = 4;
+    static final String BATCH_RECEIPT_FAMILY = "_batch_receipt";
 
     private static final String INTEGER_CODEC = " CODEC(Delta, ZSTD(3))";
     private static final String VALUE_CODEC = " CODEC(ZSTD(3))";
     private static final String SPARSE_SERIALIZATION_SETTING = "ratio_of_defaults_for_sparse_serialization=0.9";
     private static final String PURGEABLE_FAMILIES = purgeableFamilies();
-    private static final String EVENT_PARTITION_KEY = "if(family IN (" + PURGEABLE_FAMILIES + "),toYYYYMM(toDateTime(time,'UTC')),0)";
-    private static final String EVENT_SORTING_KEY = "(family,wid,x,z,if(family IN ('database_lock','user','version'),0,time),rowid)";
+    private static final String EVENT_PARTITION_KEY = "if(family='" + BATCH_RECEIPT_FAMILY + "',wid,if(family IN (" + PURGEABLE_FAMILIES
+            + "),toYYYYMM(toDateTime(time,'UTC')),0))";
+    private static final String EVENT_SORTING_KEY = "(family,wid,x,z,if(family IN ('database_lock','user','version'),0,time),rowid,if(family='"
+            + BATCH_RECEIPT_FAMILY + "',toString(batch_id),''))";
     private static final String[][] EVENT_DATA_SKIPPING_INDEX_DEFINITIONS = {
             { "batch_sequence_idx", "batch_sequence", "minmax", "1" },
             { "rowid_idx", "rowid", "bloom_filter(0.01)", "1" },
@@ -31,11 +36,10 @@ public final class ClickHouseSchema {
             { "schema_version", "UInt32" + INTEGER_CODEC },
             { "created_at", "DateTime64(3, 'UTC')" + INTEGER_CODEC }
     };
-    private static final String[][] WRITER_REGISTRATION_COLUMN_DEFINITIONS = {
-            { "dataset_id", "UUID" },
-            { "writer_id", "UUID" },
-            { "registration_order", "UInt64 DEFAULT generateSnowflakeID()" },
-            { "registered_at", "DateTime64(3, 'UTC')" }
+    private static final String[][] IDENTITY_RESERVATION_COLUMN_DEFINITIONS = {
+            { "sequence", "String" + VALUE_CODEC },
+            { "block_start", "UInt64" + INTEGER_CODEC },
+            { "writer_id", "UUID" + VALUE_CODEC }
     };
     private static final String[][] RETENTION_HIGH_WATER_COLUMN_DEFINITIONS = {
             { "batch_sequence", "UInt64" + INTEGER_CODEC },
@@ -118,11 +122,19 @@ public final class ClickHouseSchema {
         throw new IllegalStateException("Schema class");
     }
 
+    static int eventPartitionId(ClickHouseFamily family, int time) {
+        if (!family.isPurgeable()) {
+            return 0;
+        }
+        LocalDateTime timestamp = LocalDateTime.ofEpochSecond(Integer.toUnsignedLong(time), 0, ZoneOffset.UTC);
+        return timestamp.getYear() * 100 + timestamp.getMonthValue();
+    }
+
     public static List<String> createStatements(String database, String prefix) {
         Names names = new Names(database, prefix);
         List<String> statements = new ArrayList<>();
         statements.add(createStorageMetadata(names));
-        statements.add(createWriterRegistration(names));
+        statements.add(createIdentityReservation(names));
         statements.add(createRetentionHighWater(names));
         statements.add(createEventData(names));
         addCompatibilityViews(statements, names);
@@ -133,8 +145,8 @@ public final class ClickHouseSchema {
         Names names = new Names(database, prefix);
         validateTable(connection, database, names.rawTable("storage_metadata"), "MergeTree", "tuple()", "", STORAGE_METADATA_COLUMN_DEFINITIONS,
                 "fsync_after_insert=1", "fsync_part_directory=1");
-        validateTable(connection, database, names.rawTable("writer_registration"), "MergeTree", "(registration_order,writer_id)", "", WRITER_REGISTRATION_COLUMN_DEFINITIONS,
-                "fsync_after_insert=1", "fsync_part_directory=1");
+        validateTable(connection, database, names.rawTable("identity_reservation"), "MergeTree", "(sequence,block_start,writer_id)", "",
+                IDENTITY_RESERVATION_COLUMN_DEFINITIONS, "fsync_after_insert=1", "fsync_part_directory=1", "enable_block_number_column=1");
         validateTable(connection, database, names.rawTable("retention_high_water"), "MergeTree", "(batch_sequence,family,rowid)", "", RETENTION_HIGH_WATER_COLUMN_DEFINITIONS,
                 "fsync_after_insert=1", "fsync_part_directory=1", "non_replicated_deduplication_window=1000");
         validateTable(connection, database, names.rawTable("event_data"), "CoalescingMergeTree", EVENT_SORTING_KEY, EVENT_PARTITION_KEY, EVENT_COLUMN_DEFINITIONS,
@@ -144,7 +156,7 @@ public final class ClickHouseSchema {
 
     static void requireUnownedTablesEmpty(Connection connection, String database, String prefix) throws SQLException {
         Names names = new Names(database, prefix);
-        for (String suffix : new String[] { "writer_registration", "retention_high_water", "event_data" }) {
+        for (String suffix : new String[] { "identity_reservation", "retention_high_water", "event_data" }) {
             try (PreparedStatement statement = connection.prepareStatement("SELECT 1 FROM " + names.table(suffix) + " LIMIT 1"); ResultSet resultSet = statement.executeQuery()) {
                 if (resultSet.next()) {
                     throw new SQLException("ClickHouse table " + names.rawTable(suffix) + " contains data without CoreProtect storage metadata");
@@ -219,11 +231,11 @@ public final class ClickHouseSchema {
                 + " SETTINGS fsync_after_insert=1,fsync_part_directory=1";
     }
 
-    private static String createWriterRegistration(Names names) {
-        return table(names.writerRegistration, columnDefinitions(WRITER_REGISTRATION_COLUMN_DEFINITIONS))
+    private static String createIdentityReservation(Names names) {
+        return table(names.identityReservation, columnDefinitions(IDENTITY_RESERVATION_COLUMN_DEFINITIONS))
                 + " ENGINE = MergeTree"
-                + " ORDER BY (registration_order,writer_id)"
-                + " SETTINGS fsync_after_insert=1,fsync_part_directory=1";
+                + " ORDER BY (sequence,block_start,writer_id)"
+                + " SETTINGS fsync_after_insert=1,fsync_part_directory=1,enable_block_number_column=1";
     }
 
     private static String createRetentionHighWater(Names names) {
@@ -271,7 +283,7 @@ public final class ClickHouseSchema {
     }
 
     private static void addCompatibilityViews(List<String> statements, Names names) {
-        statements.add(view(names, ClickHouseFamily.ART_MAP, "e.rowid AS rowid,e.id AS id,e.name AS art"));
+        statements.add(currentView(names, ClickHouseFamily.ART_MAP, "e.rowid AS rowid,e.id AS id,e.name AS art"));
         statements.add(rollbackView(names, ClickHouseFamily.BLOCK, "e.rowid AS rowid,e.time AS time,e.user_id AS `user`," + location("wid") + "," + location("x") + ",e.y AS y," + location("z") + ",e.type AS type,e.data AS data," + binary("e.meta", "meta") + "," + binary("e.blockdata", "blockdata") + ",e.action AS action"));
         statements.add(view(names, ClickHouseFamily.CHAT, "e.rowid AS rowid,e.time AS time,e.user_id AS `user`," + location("wid") + "," + location("x") + ",e.y AS y," + location("z") + ",e.message AS message"));
         statements.add(view(names, ClickHouseFamily.COMMAND, "e.rowid AS rowid,e.time AS time,e.user_id AS `user`," + location("wid") + "," + location("x") + ",e.y AS y," + location("z") + ",e.message AS message"));
@@ -282,16 +294,16 @@ public final class ClickHouseSchema {
         statements.add(currentView(names, ClickHouseFamily.DATABASE_LOCK, "e.rowid AS rowid,e.status AS status,e.database_lock_time AS time"));
         statements.add(view(names, ClickHouseFamily.ENTITY, "e.rowid AS rowid,e.time AS time," + binary("e.payload", "data")));
         statements.add(entitySpawnView(names));
-        statements.add(view(names, ClickHouseFamily.ENTITY_MAP, "e.rowid AS rowid,e.id AS id,e.name AS entity"));
-        statements.add(view(names, ClickHouseFamily.MATERIAL_MAP, "e.rowid AS rowid,e.id AS id,e.name AS material"));
-        statements.add(view(names, ClickHouseFamily.BLOCKDATA_MAP, "e.rowid AS rowid,e.id AS id,e.text AS data"));
+        statements.add(currentView(names, ClickHouseFamily.ENTITY_MAP, "e.rowid AS rowid,e.id AS id,e.name AS entity"));
+        statements.add(currentView(names, ClickHouseFamily.MATERIAL_MAP, "e.rowid AS rowid,e.id AS id,e.name AS material"));
+        statements.add(currentView(names, ClickHouseFamily.BLOCKDATA_MAP, "e.rowid AS rowid,e.id AS id,e.text AS data"));
         statements.add(view(names, ClickHouseFamily.SESSION, "e.rowid AS rowid,e.time AS time,e.user_id AS `user`," + location("wid") + "," + location("x") + ",e.y AS y," + location("z") + ",e.action AS action"));
         statements.add(view(names, ClickHouseFamily.SIGN, "e.rowid AS rowid,e.time AS time,e.user_id AS `user`," + location("wid") + "," + location("x") + ",e.y AS y," + location("z") + ",e.action AS action,e.color AS color,e.color_secondary AS color_secondary,e.sign_data AS data,e.waxed AS waxed,e.face AS face,e.line_1 AS line_1,e.line_2 AS line_2,e.line_3 AS line_3,e.line_4 AS line_4,e.line_5 AS line_5,e.line_6 AS line_6,e.line_7 AS line_7,e.line_8 AS line_8"));
         statements.add(view(names, ClickHouseFamily.SKULL, "e.rowid AS rowid,e.time AS time,e.name AS owner,e.text AS skin"));
-        statements.add(currentView(names, ClickHouseFamily.USER, "e.rowid AS rowid,e.time AS time,e.user_name AS `user`,e.uuid AS uuid"));
+        statements.add(currentView(names, ClickHouseFamily.USER, "e.rowid AS rowid,e.time AS time,e.user_name AS `user`,ifNull(e.uuid,'') AS uuid"));
         statements.add(view(names, ClickHouseFamily.USERNAME_LOG, "e.rowid AS rowid,e.time AS time,e.uuid AS uuid,e.user_name AS `user`"));
         statements.add(currentView(names, ClickHouseFamily.VERSION, "e.rowid AS rowid,e.time AS time,e.version AS version"));
-        statements.add(view(names, ClickHouseFamily.WORLD, "e.rowid AS rowid,e.id AS id,e.name AS world"));
+        statements.add(currentView(names, ClickHouseFamily.WORLD, "e.rowid AS rowid,e.id AS id,e.name AS world"));
     }
 
     private static String view(Names names, ClickHouseFamily family, String projection) {
@@ -466,7 +478,7 @@ public final class ClickHouseSchema {
         private final String eventData;
         private final String retentionHighWater;
         private final String storageMetadata;
-        private final String writerRegistration;
+        private final String identityReservation;
 
         private Names(String database, String prefix) {
             this.database = ClickHouseIdentifiers.quote(database, "ClickHouse database");
@@ -474,7 +486,7 @@ public final class ClickHouseSchema {
             eventData = table("event_data");
             retentionHighWater = table("retention_high_water");
             storageMetadata = table("storage_metadata");
-            writerRegistration = table("writer_registration");
+            identityReservation = table("identity_reservation");
         }
 
         private String table(String suffix) {

src/main/java/net/coreprotect/database/clickhouse/ClickHouseStartupReconciler.java

@@ -35,8 +35,8 @@ public final class ClickHouseStartupReconciler {
 
     private static Map<ClickHouseFamily, Long> readRawRowIds(Connection connection, String eventTable, String retentionHighWater) throws SQLException {
         String sql = "SELECT family,max(rowid) FROM ("
-                + "SELECT family,rowid FROM " + eventTable + " UNION ALL "
-                + "SELECT family,rowid FROM " + retentionHighWater + ") GROUP BY family";
+                + "SELECT family,rowid FROM " + eventTable + " WHERE family!='" + ClickHouseSchema.BATCH_RECEIPT_FAMILY + "' UNION ALL "
+                + "SELECT family,rowid FROM " + retentionHighWater + " WHERE family!='" + ClickHouseSchema.BATCH_RECEIPT_FAMILY + "') GROUP BY family";
         Map<ClickHouseFamily, Long> rowIds = new EnumMap<>(ClickHouseFamily.class);
         try (PreparedStatement statement = connection.prepareStatement(sql)) {
             try (ResultSet resultSet = statement.executeQuery()) {

src/main/java/net/coreprotect/database/clickhouse/ClickHouseStateBatch.java

@@ -44,18 +44,19 @@ final class ClickHouseStateBatch implements AutoCloseable {
         entityStateUpdates.put(pointer.getRowId(), state);
     }
 
-    void appendTo(ClickHouseRowBinaryBuffer rows, int firstOrdinal) throws SQLException {
+    void appendTo(ClickHouseRowBinaryBuffer rows, int firstOrdinal, Map<Integer, Integer> partitionRowCounts) throws SQLException {
         if (sealed) {
             throw new IllegalStateException("ClickHouse state batch is already appended");
         }
         ensureWritable();
         Objects.requireNonNull(rows, "rows");
+        Objects.requireNonNull(partitionRowCounts, "partitionRowCounts");
         int ordinal = firstOrdinal;
         for (Map<Long, RollbackUpdate> familyUpdates : rollbackUpdates.values()) {
             for (RollbackUpdate update : familyUpdates.values()) {
                 beginSparseRow(rows, update, ordinal++);
                 rows.set("rolled_back", update.rolledBack);
-                rows.commitRow("rollback state update");
+                commitRow(rows, "rollback state update", update.family, update.time, partitionRowCounts);
             }
         }
         for (ClickHouseEntityState state : entityStateUpdates.values()) {
@@ -78,7 +79,7 @@ final class ClickHouseStateBatch implements AutoCloseable {
             rows.set("entity_data", data);
             rows.set("entity_data_present", data == null ? 0 : 1);
             rows.set("removed", state.isRemoved() ? 1 : 0);
-            rows.commitRow("entity state update");
+            commitRow(rows, "entity state update", pointer.getFamily(), pointer.getTime(), partitionRowCounts);
         }
         sealed = true;
     }
@@ -167,6 +168,13 @@ final class ClickHouseStateBatch implements AutoCloseable {
         rows.set("z", z);
     }
 
+    private static void commitRow(ClickHouseRowBinaryBuffer rows, String description, ClickHouseFamily family, int time,
+            Map<Integer, Integer> partitionRowCounts) throws SQLException {
+        int partitionId = ClickHouseSchema.eventPartitionId(family, time);
+        rows.commitRow(description, partitionId);
+        partitionRowCounts.merge(partitionId, 1, Math::addExact);
+    }
+
     private void ensureWritable() {
         if (sealed) {
             throw new IllegalStateException("ClickHouse state batch is sealed");

src/main/java/net/coreprotect/database/clickhouse/ClickHouseWriteBatch.java

@@ -2,13 +2,17 @@ package net.coreprotect.database.clickhouse;
 
 import java.io.InputStream;
 import java.sql.SQLException;
+import java.util.HashSet;
+import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
 
 public final class ClickHouseWriteBatch implements AutoCloseable {
 
     private final ClickHouseBatchIdentity identity;
     private final ClickHouseEventBatch events;
     private final ClickHouseStateBatch state;
+    private final Set<Integer> publishedPartitions = new HashSet<>();
     private ClickHouseBatchReceipt receipt;
     private boolean closed;
     private boolean published;
@@ -45,12 +49,12 @@ public final class ClickHouseWriteBatch implements AutoCloseable {
             return receipt;
         }
         int eventCount = events.size();
-        events.seal(state);
         int rollbackCount = state.getRollbackCount();
         int entityStateCount = state.getEntityStateCount();
         int rowCount = Math.addExact(eventCount, Math.addExact(rollbackCount, entityStateCount));
         int logicalRowCount = Math.subtractExact(Math.addExact(events.logicalSize(), Math.addExact(rollbackCount, entityStateCount)), state.getLocalOverlapCount(eventCount));
-        receipt = new ClickHouseBatchReceipt(identity.getBatchSequence(), identity.getBatchId(), rowCount, logicalRowCount);
+        Map<Integer, Integer> partitionRowCounts = events.seal(state);
+        receipt = new ClickHouseBatchReceipt(identity.getBatchSequence(), identity.getBatchId(), rowCount, logicalRowCount, partitionRowCounts);
         return receipt;
     }
 
@@ -61,9 +65,25 @@ public final class ClickHouseWriteBatch implements AutoCloseable {
 
     void markPublished() {
         ensureOpen();
+        if (receipt == null || !publishedPartitions.containsAll(receipt.getPartitionRowCounts().keySet())) {
+            throw new IllegalStateException("ClickHouse batch has unpublished partitions");
+        }
         published = true;
     }
 
+    boolean isPartitionPublished(int partitionId) {
+        ensureOpen();
+        return publishedPartitions.contains(partitionId);
+    }
+
+    void markPartitionPublished(int partitionId) {
+        ensureOpen();
+        if (receipt == null || !receipt.getPartitionRowCounts().containsKey(partitionId)) {
+            throw new IllegalArgumentException("ClickHouse batch does not contain partition " + partitionId);
+        }
+        publishedPartitions.add(partitionId);
+    }
+
     ClickHouseBatchIdentity getIdentity() {
         return identity;
     }
@@ -80,6 +100,13 @@ public final class ClickHouseWriteBatch implements AutoCloseable {
         return events.openRows();
     }
 
+    InputStream openRows(int partitionId) {
+        if (receipt == null) {
+            throw new IllegalStateException("ClickHouse write batch is not sealed");
+        }
+        return events.openRows(partitionId);
+    }
+
     Checkpoint checkpoint() {
         ensureOpen();
         if (receipt != null) {
@@ -104,6 +131,7 @@ public final class ClickHouseWriteBatch implements AutoCloseable {
             closed = true;
             events.close();
             state.close();
+            publishedPartitions.clear();
         }
     }
 

src/main/java/net/coreprotect/database/clickhouse/ClickHouseWriterRegistration.java

@@ -1,99 +1,56 @@
 package net.coreprotect.database.clickhouse;
 
 import java.io.IOException;
-import java.nio.ByteBuffer;
 import java.nio.channels.FileChannel;
 import java.nio.channels.FileLock;
 import java.nio.channels.OverlappingFileLockException;
-import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.StandardOpenOption;
-import java.sql.Connection;
-import java.sql.PreparedStatement;
-import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.Objects;
-import java.util.UUID;
 
 final class ClickHouseWriterRegistration implements AutoCloseable {
 
     private static final String WRITER_FILE = ".clickhouse-writer";
-    private static final int MAX_WRITER_ID_BYTES = 64;
 
-    private final ClickHouseJdbc jdbc;
-    private final String table;
-    private final UUID datasetId;
     private final Path writerFile;
-    private UUID writerId;
     private FileChannel writerChannel;
     private FileLock writerLock;
-    private boolean owned;
     private boolean closed;
 
-    ClickHouseWriterRegistration(ClickHouseJdbc jdbc, String database, String prefix, UUID datasetId, Path controlDirectory) {
-        this.jdbc = Objects.requireNonNull(jdbc, "jdbc");
-        table = ClickHouseIdentifiers.qualified(database, prefix + "writer_registration");
-        this.datasetId = Objects.requireNonNull(datasetId, "datasetId");
+    ClickHouseWriterRegistration(Path controlDirectory) {
         writerFile = Objects.requireNonNull(controlDirectory, "controlDirectory").resolve(WRITER_FILE);
     }
 
     synchronized void acquire() throws SQLException {
         ensureOpen();
-        if (owned) {
+        if (writerLock != null && writerLock.isValid()) {
             return;
         }
-        acquireLocalWriter();
-        Registration registration = read();
-        if (registration == null) {
+        try {
+            Files.createDirectories(writerFile.toAbsolutePath().getParent());
+            writerChannel = FileChannel.open(writerFile, StandardOpenOption.CREATE, StandardOpenOption.READ, StandardOpenOption.WRITE);
             try {
-                insert();
+                writerLock = writerChannel.tryLock();
             }
-            catch (SQLException insertionFailure) {
-                try {
-                    registration = read();
-                }
-                catch (SQLException reconciliationFailure) {
-                    insertionFailure.addSuppressed(reconciliationFailure);
-                    throw insertionFailure;
-                }
-                if (registration == null) {
-                    throw insertionFailure;
-                }
+            catch (OverlappingFileLockException exception) {
+                throw new SQLException("This CoreProtect installation already has an active ClickHouse writer", exception);
             }
-            if (registration == null) {
-                registration = read();
+            if (writerLock == null) {
+                throw new SQLException("This CoreProtect installation already has an active ClickHouse writer");
             }
         }
-        if (registration == null) {
-            throw new SQLException("ClickHouse writer registration is not visible after creation");
+        catch (IOException exception) {
+            throw new SQLException("Failed to open the local ClickHouse writer registration", exception);
         }
-        requireRegistration(registration);
-        owned = true;
     }
 
     synchronized void verifyOwned() throws SQLException {
         ensureOpen();
-        requireOwned();
-        Registration registration = read();
-        if (registration == null) {
-            throw new OwnershipException("ClickHouse writer registration is not visible");
+        if (writerLock == null || !writerLock.isValid()) {
+            throw new OwnershipException("ClickHouse writer registration is not owned by this CoreProtect installation");
         }
-        requireRegistration(registration);
-    }
-
-    private void requireRegistration(Registration registration) throws SQLException {
-        if (!datasetId.equals(registration.datasetId)) {
-            throw new OwnershipException("ClickHouse writer registration belongs to a different storage identity");
-        }
-        if (!writerId.equals(registration.writerId)) {
-            throw new OwnershipException("ClickHouse dataset is registered to a different CoreProtect installation");
-        }
-    }
-
-    synchronized void ensureOwned() throws SQLException {
-        ensureOpen();
-        requireOwned();
     }
 
     @Override
@@ -126,7 +83,6 @@ final class ClickHouseWriterRegistration implements AutoCloseable {
         finally {
             writerLock = null;
             writerChannel = null;
-            owned = false;
             closed = true;
         }
         if (failure != null) {
@@ -134,113 +90,12 @@ final class ClickHouseWriterRegistration implements AutoCloseable {
         }
     }
 
-    private void acquireLocalWriter() throws SQLException {
-        if (writerLock != null && writerLock.isValid()) {
-            return;
-        }
-        try {
-            Files.createDirectories(writerFile.toAbsolutePath().getParent());
-            writerChannel = FileChannel.open(writerFile, StandardOpenOption.CREATE, StandardOpenOption.READ, StandardOpenOption.WRITE);
-            try {
-                writerLock = writerChannel.tryLock();
-            }
-            catch (OverlappingFileLockException exception) {
-                throw new SQLException("This CoreProtect installation already has an active ClickHouse writer", exception);
-            }
-            if (writerLock == null) {
-                throw new SQLException("This CoreProtect installation already has an active ClickHouse writer");
-            }
-            writerId = readOrCreateWriterId();
-        }
-        catch (IOException exception) {
-            throw new SQLException("Failed to open the local ClickHouse writer registration", exception);
-        }
-    }
-
-    private UUID readOrCreateWriterId() throws IOException, SQLException {
-        long size = writerChannel.size();
-        if (size == 0) {
-            UUID id = UUID.randomUUID();
-            byte[] encoded = id.toString().getBytes(StandardCharsets.US_ASCII);
-            writerChannel.position(0);
-            writeFully(writerChannel, ByteBuffer.wrap(encoded));
-            writerChannel.truncate(encoded.length);
-            writerChannel.force(true);
-            return id;
-        }
-        if (size > MAX_WRITER_ID_BYTES) {
-            throw new SQLException("Local ClickHouse writer registration is invalid");
-        }
-        ByteBuffer encoded = ByteBuffer.allocate((int) size);
-        writerChannel.position(0);
-        while (encoded.hasRemaining()) {
-            if (writerChannel.read(encoded) < 0) {
-                throw new SQLException("Local ClickHouse writer registration is incomplete");
-            }
-        }
-        String value = new String(encoded.array(), StandardCharsets.US_ASCII).trim();
-        try {
-            UUID id = UUID.fromString(value);
-            if (!id.toString().equalsIgnoreCase(value)) {
-                throw new IllegalArgumentException("Non-canonical UUID");
-            }
-            return id;
-        }
-        catch (IllegalArgumentException exception) {
-            throw new SQLException("Local ClickHouse writer registration is invalid", exception);
-        }
-    }
-
-    private Registration read() throws SQLException {
-        String sql = "SELECT dataset_id,writer_id FROM " + table
-                + " ORDER BY registration_order,writer_id LIMIT 1";
-        try (Connection connection = jdbc.openConnection(); PreparedStatement statement = connection.prepareStatement(sql); ResultSet resultSet = statement.executeQuery()) {
-            if (!resultSet.next()) {
-                return null;
-            }
-            return new Registration(UUID.fromString(resultSet.getString(1)), UUID.fromString(resultSet.getString(2)));
-        }
-    }
-
-    private void insert() throws SQLException {
-        String sql = "INSERT INTO " + table
-                + " (dataset_id,writer_id,registered_at) VALUES (?,?,now64(3, 'UTC'))";
-        try (Connection connection = jdbc.openConnection(); PreparedStatement statement = connection.prepareStatement(sql)) {
-            statement.setObject(1, datasetId);
-            statement.setObject(2, writerId);
-            statement.execute();
-        }
-    }
-
-    private void requireOwned() throws SQLException {
-        if (!owned || writerLock == null || !writerLock.isValid()) {
-            throw new OwnershipException("ClickHouse writer registration is not owned by this CoreProtect installation");
-        }
-    }
-
     private void ensureOpen() {
         if (closed) {
             throw new IllegalStateException("ClickHouse writer registration is closed");
         }
     }
 
-    private static void writeFully(FileChannel channel, ByteBuffer data) throws IOException {
-        while (data.hasRemaining()) {
-            channel.write(data);
-        }
-    }
-
-    private static final class Registration {
-
-        private final UUID datasetId;
-        private final UUID writerId;
-
-        private Registration(UUID datasetId, UUID writerId) {
-            this.datasetId = datasetId;
-            this.writerId = writerId;
-        }
-    }
-
     static final class OwnershipException extends SQLException {
 
         private static final long serialVersionUID = 1L;

src/main/java/net/coreprotect/database/logger/PlayerKillLogger.java

@@ -1,7 +1,5 @@
 package net.coreprotect.database.logger;
 
-import java.util.Locale;
-
 import org.bukkit.Bukkit;
 import org.bukkit.Location;
 import org.bukkit.entity.EntityType;
@@ -29,9 +27,7 @@ public class PlayerKillLogger {
                 return;
             }
 
-            if (ConfigHandler.playerIdCache.get(player.toLowerCase(Locale.ROOT)) == null) {
-                preparedStmt.resolveUserId(player, null);
-            }
+            int playerId = UserStatement.getId(preparedStmt, player, true);
 
             Location initialLocation = location.clone();
             CoreProtectPreLogEvent event = new CoreProtectPreLogEvent(user, initialLocation, CoreProtectPreLogEvent.Action.PLAYER_KILL, LookupActions.ENTITY_KILL, null, EntityType.PLAYER, null);
@@ -44,7 +40,6 @@ public class PlayerKillLogger {
             }
 
             int userId = UserStatement.getId(preparedStmt, event.getUser(), true);
-            int playerId = ConfigHandler.playerIdCache.get(player.toLowerCase(Locale.ROOT));
             Location eventLocation = event.getLocation();
             int wid = WorldUtils.getWorldId(eventLocation.getWorld().getName());
             int time = (int) (System.currentTimeMillis() / 1000L);

src/main/java/net/coreprotect/database/lookup/BlockLookup.java

@@ -77,7 +77,7 @@ public class BlockLookup {
                     count = results.getInt("count");
                 }
                 results.close();
-                query = "SELECT time," + ConfigHandler.databaseType.getUserColumn() + ",action,type,data,rolled_back FROM " + ConfigHandler.prefix + "block " + WorldUtils.getWidIndex("block") + "WHERE " + where + " ORDER BY rowid DESC LIMIT " + limit + " OFFSET " + page_start;
+                query = "SELECT time," + ConfigHandler.databaseType.getUserColumn() + ",action,type,data,rolled_back FROM " + ConfigHandler.prefix + "block " + WorldUtils.getWidIndex("block") + "WHERE " + where + " ORDER BY " + ConfigHandler.getDescendingEventOrder() + " LIMIT " + limit + " OFFSET " + page_start;
                 results = statement.executeQuery(query);
             }
 

src/main/java/net/coreprotect/database/lookup/ChestTransactionLookup.java

@@ -83,7 +83,7 @@ public class ChestTransactionLookup {
             String tableName = "container";
             String where = "wid = " + worldId + " AND (x = " + x + " OR x = " + x2 + ") AND (z = " + z + " OR z = " + z2 + ") AND y = " + y;
             String index = WorldUtils.getWidIndex("container");
-            String order = "rowid DESC";
+            String order = ConfigHandler.getDescendingEventOrder();
             if (entitySpawnRowId != null) {
                 table = ConfigHandler.prefix + "entity_container";
                 tableName = "entity_container";

src/main/java/net/coreprotect/database/lookup/InteractionLookup.java

@@ -75,7 +75,7 @@ public class InteractionLookup {
                     count = results.getInt("count");
                 }
                 results.close();
-                query = "SELECT time," + ConfigHandler.databaseType.getUserColumn() + ",action,type,data,rolled_back FROM " + ConfigHandler.prefix + "block " + WorldUtils.getWidIndex("block") + "WHERE " + where + " ORDER BY rowid DESC LIMIT " + limit + " OFFSET " + pageStart;
+                query = "SELECT time," + ConfigHandler.databaseType.getUserColumn() + ",action,type,data,rolled_back FROM " + ConfigHandler.prefix + "block " + WorldUtils.getWidIndex("block") + "WHERE " + where + " ORDER BY " + ConfigHandler.getDescendingEventOrder() + " LIMIT " + limit + " OFFSET " + pageStart;
                 results = statement.executeQuery(query);
             }
 

src/main/java/net/coreprotect/database/lookup/SignMessageLookup.java

@@ -76,7 +76,7 @@ public class SignMessageLookup {
                     count = results.getInt("count");
                 }
                 results.close();
-                query = "SELECT time," + ConfigHandler.databaseType.getUserColumn() + ",face,line_1,line_2,line_3,line_4,line_5,line_6,line_7,line_8 FROM " + ConfigHandler.prefix + "sign " + WorldUtils.getWidIndex("sign") + "WHERE " + where + " ORDER BY rowid DESC LIMIT " + limit + " OFFSET " + pageStart;
+                query = "SELECT time," + ConfigHandler.databaseType.getUserColumn() + ",face,line_1,line_2,line_3,line_4,line_5,line_6,line_7,line_8 FROM " + ConfigHandler.prefix + "sign " + WorldUtils.getWidIndex("sign") + "WHERE " + where + " ORDER BY " + ConfigHandler.getDescendingEventOrder() + " LIMIT " + limit + " OFFSET " + pageStart;
                 results = statement.executeQuery(query);
             }
 

src/main/java/net/coreprotect/database/rollback/Rollback.java

@@ -226,6 +226,7 @@ public class Rollback extends RollbackUtil {
             TreeMap<Long, Integer> chunkList = new TreeMap<>();
             HashMap<Integer, HashMap<Long, ArrayList<Object[]>>> dataList = new HashMap<>();
             HashMap<Integer, HashMap<Long, ArrayList<Object[]>>> itemDataList = new HashMap<>();
+            Set<Integer> inventoryUserUuids = new HashSet<>();
             int listC = 0;
             while (listC < 2) {
                 List<Object[]> scanList = rollbackLookupList;
@@ -253,7 +254,10 @@ public class Rollback extends RollbackUtil {
                         chunkList.put(chunkKey, distance);
                     }
 
-                    UserStatement.getName(statement.getConnection(), userId);
+                    String rowUser = UserStatement.getName(statement.getConnection(), userId);
+                    if (inventoryRollback && inventoryUserUuids.add(userId) && rowUser != null && !rowUser.isEmpty()) {
+                        UserStatement.getUuid(statement.getConnection(), rowUser);
+                    }
 
                     HashMap<Integer, HashMap<Long, ArrayList<Object[]>>> modifyList = dataList;
                     if (listC == 1) {

src/main/java/net/coreprotect/database/statement/UserStatement.java

@@ -178,12 +178,15 @@ public class UserStatement {
 
             ConfigHandler.playerIdCacheReversed.put(id, user);
             ConfigHandler.playerIdCache.put(user.toLowerCase(Locale.ROOT), id);
-            if (uuid != null) {
+            if (uuid != null && !uuid.isEmpty()) {
                 ConfigHandler.uuidCache.put(user.toLowerCase(Locale.ROOT), uuid);
                 ConfigHandler.uuidCacheReversed.put(uuid, user);
             }
         }
         catch (Exception e) {
+            if (ConfigHandler.databaseType.isClickHouse()) {
+                throw new IllegalStateException("Unable to resolve ClickHouse user identifier " + id, e);
+            }
             ErrorReporter.report(e);
         }
 
@@ -198,17 +201,72 @@ public class UserStatement {
         return user;
     }
 
-    public static String getNameByUuid(String uuid) {
-        return ConfigHandler.uuidCacheReversed.get(uuid);
+    public static String getNameByUuid(Connection connection, String uuid) {
+        if (uuid == null || uuid.isEmpty()) {
+            return null;
+        }
+        String user = ConfigHandler.uuidCacheReversed.get(uuid);
+        if (!ConfigHandler.databaseType.isClickHouse() || user != null) {
+            return user;
+        }
+
+        try {
+            String query = "SELECT rowid," + ConfigHandler.databaseType.getUserColumn() + " FROM " + ConfigHandler.prefix + "user WHERE uuid=? ORDER BY rowid LIMIT 1";
+            try (PreparedStatement statement = connection.prepareStatement(query)) {
+                statement.setString(1, uuid);
+                try (ResultSet resultSet = statement.executeQuery()) {
+                    if (!resultSet.next()) {
+                        return null;
+                    }
+                    int id = resultSet.getInt("rowid");
+                    String resolvedUser = resultSet.getString("user");
+                    if (resolvedUser == null || resolvedUser.isEmpty()) {
+                        return null;
+                    }
+                    String lowerUser = resolvedUser.toLowerCase(Locale.ROOT);
+                    ConfigHandler.playerIdCache.put(lowerUser, id);
+                    ConfigHandler.playerIdCacheReversed.put(id, resolvedUser);
+                    ConfigHandler.uuidCache.put(lowerUser, uuid);
+                    ConfigHandler.uuidCacheReversed.put(uuid, resolvedUser);
+                    return resolvedUser;
+                }
+            }
+        }
+        catch (SQLException e) {
+            throw new IllegalStateException("Unable to resolve ClickHouse user UUID " + uuid, e);
+        }
     }
 
     public static String getUuid(Connection connection, String user) throws SQLException {
         String lowerUser = user.toLowerCase(Locale.ROOT);
         String uuid = ConfigHandler.uuidCache.get(lowerUser);
-        if (uuid == null && findId(connection, user) > -1) {
-            uuid = ConfigHandler.uuidCache.get(lowerUser);
+        if (uuid != null && (!ConfigHandler.databaseType.isClickHouse() || !uuid.isEmpty())) {
+            return uuid;
         }
-        return uuid;
+        int id = findId(connection, user);
+        if (id < 1) {
+            return null;
+        }
+        if (ConfigHandler.databaseType.isClickHouse()) {
+            uuid = ConfigHandler.uuidCache.get(lowerUser);
+            if (uuid != null && !uuid.isEmpty()) {
+                return uuid;
+            }
+            String resolvedUser = loadName(connection, id);
+            if (resolvedUser.isEmpty()) {
+                throw new SQLException("ClickHouse user identifier " + id + " has no user row");
+            }
+            uuid = ConfigHandler.uuidCache.get(resolvedUser.toLowerCase(Locale.ROOT));
+            return uuid == null || uuid.isEmpty() ? null : uuid;
+        }
+        return ConfigHandler.uuidCache.get(lowerUser);
+    }
+
+    public static void clearClickHouseCaches() {
+        ConfigHandler.playerIdCache.clear();
+        ConfigHandler.playerIdCacheReversed.clear();
+        ConfigHandler.uuidCache.clear();
+        ConfigHandler.uuidCacheReversed.clear();
     }
 
 }

src/main/java/net/coreprotect/utility/EntityUtils.java

@@ -31,6 +31,9 @@ public class EntityUtils extends Queue {
         int id = -1;
         name = name.toLowerCase(Locale.ROOT).trim();
 
+        if (ConfigHandler.databaseType.isClickHouse()) {
+            return ConfigHandler.resolveIdentifierId(ConfigHandler.CacheType.ENTITIES, name, internal);
+        }
         if (ConfigHandler.entities.get(name) != null) {
             id = ConfigHandler.entities.get(name);
         }
@@ -95,6 +98,10 @@ public class EntityUtils extends Queue {
 
     public static String getEntityName(int id) {
         // Internal ID pulled from DB
+        if (ConfigHandler.databaseType.isClickHouse()) {
+            String entityName = ConfigHandler.getIdentifierValue(ConfigHandler.CacheType.ENTITIES, id);
+            return entityName == null ? "" : entityName;
+        }
         String entityName = "";
         String cachedName = ConfigHandler.entitiesReversed.get(id);
         if (cachedName != null) {

src/main/java/net/coreprotect/utility/MaterialUtils.java

@@ -41,6 +41,9 @@ public class MaterialUtils extends Queue {
             name = NAMESPACE + name;
         }
 
+        if (ConfigHandler.databaseType.isClickHouse()) {
+            return ConfigHandler.resolveIdentifierId(ConfigHandler.CacheType.MATERIALS, name, internal);
+        }
         if (ConfigHandler.materials.get(name) != null) {
             id = ConfigHandler.materials.get(name);
         }
@@ -66,6 +69,9 @@ public class MaterialUtils extends Queue {
         int id = -1;
         data = data.toLowerCase(Locale.ROOT).trim();
 
+        if (ConfigHandler.databaseType.isClickHouse()) {
+            return ConfigHandler.resolveIdentifierId(ConfigHandler.CacheType.BLOCKDATA, data, internal);
+        }
         if (ConfigHandler.blockdata.get(data) != null) {
             id = ConfigHandler.blockdata.get(data);
         }
@@ -89,6 +95,10 @@ public class MaterialUtils extends Queue {
 
     public static String getBlockDataString(int id) {
         // Internal ID pulled from DB
+        if (ConfigHandler.databaseType.isClickHouse()) {
+            String blockdata = ConfigHandler.getIdentifierValue(ConfigHandler.CacheType.BLOCKDATA, id);
+            return blockdata == null ? "" : blockdata;
+        }
         String blockdata = "";
         String cachedBlockdata = ConfigHandler.blockdataReversed.get(id);
         if (cachedBlockdata != null) {
@@ -98,6 +108,10 @@ public class MaterialUtils extends Queue {
     }
 
     public static String getBlockName(int id) {
+        if (ConfigHandler.databaseType.isClickHouse()) {
+            String name = ConfigHandler.getIdentifierValue(ConfigHandler.CacheType.MATERIALS, id);
+            return name == null ? "" : name;
+        }
         String name = "";
         String cachedName = ConfigHandler.materialsReversed.get(id);
         if (cachedName != null) {
@@ -165,6 +179,9 @@ public class MaterialUtils extends Queue {
         int id = -1;
         name = name.toLowerCase(Locale.ROOT).trim();
 
+        if (ConfigHandler.databaseType.isClickHouse()) {
+            return ConfigHandler.resolveIdentifierId(ConfigHandler.CacheType.ART, name, internal);
+        }
         if (ConfigHandler.art.get(name) != null) {
             id = ConfigHandler.art.get(name);
         }
@@ -192,6 +209,10 @@ public class MaterialUtils extends Queue {
 
     public static String getArtName(int id) {
         // Internal ID pulled from DB
+        if (ConfigHandler.databaseType.isClickHouse()) {
+            String artName = ConfigHandler.getIdentifierValue(ConfigHandler.CacheType.ART, id);
+            return artName == null ? "" : artName;
+        }
         String artname = "";
         String cachedName = ConfigHandler.artReversed.get(id);
         if (cachedName != null) {

src/main/java/net/coreprotect/utility/WorldUtils.java

@@ -13,6 +13,9 @@ public class WorldUtils extends Queue {
     }
 
     public static int getWorldId(String name) {
+        if (ConfigHandler.databaseType.isClickHouse()) {
+            return ConfigHandler.resolveIdentifierId(ConfigHandler.CacheType.WORLDS, name, true);
+        }
         int id = -1;
         try {
             if (ConfigHandler.worlds.get(name) == null) {
@@ -37,6 +40,10 @@ public class WorldUtils extends Queue {
     }
 
     public static String getWorldName(int id) {
+        if (ConfigHandler.databaseType.isClickHouse()) {
+            String name = ConfigHandler.getIdentifierValue(ConfigHandler.CacheType.WORLDS, id);
+            return name == null ? "" : name;
+        }
         String name = "";
         try {
             String cachedName = ConfigHandler.worldsReversed.get(id);