From 58db9bde803ee93baf16dc9ecd18f76d76cd5a19 Mon Sep 17 00:00:00 2001 From: Prashant Pandey Date: Tue, 21 Jul 2026 14:29:26 +0530 Subject: [PATCH 1/3] Sort keys to prevent deadlocks in updates --- .../postgres/FlatPostgresCollection.java | 19 ++++++++++++++++--- 1 file changed, 16 insertions(+), 3 deletions(-) diff --git a/document-store/src/main/java/org/hypertrace/core/documentstore/postgres/FlatPostgresCollection.java b/document-store/src/main/java/org/hypertrace/core/documentstore/postgres/FlatPostgresCollection.java index cbbcbfa3..1c41cedb 100644 --- a/document-store/src/main/java/org/hypertrace/core/documentstore/postgres/FlatPostgresCollection.java +++ b/document-store/src/main/java/org/hypertrace/core/documentstore/postgres/FlatPostgresCollection.java @@ -28,6 +28,7 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; +import java.util.Comparator; import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; @@ -364,7 +365,7 @@ public boolean bulkUpsert(Map documents) { Map parsedDocuments = new LinkedHashMap<>(); List ignoredDocuments = new ArrayList<>(); - for (Map.Entry entry : documents.entrySet()) { + for (Map.Entry entry : sortedByKey(documents)) { List skippedFields = new ArrayList<>(); TypedDocument parsed = parseDocument(entry.getValue(), tableName, skippedFields); @@ -458,7 +459,7 @@ public boolean bulkCreateOrReplace(Map documents) { Map parsedDocuments = new LinkedHashMap<>(); List ignoredDocuments = new ArrayList<>(); - for (Map.Entry entry : documents.entrySet()) { + for (Map.Entry entry : sortedByKey(documents)) { List skippedFields = new ArrayList<>(); TypedDocument parsed = parseDocument(entry.getValue(), tableName, skippedFields); @@ -665,6 +666,12 @@ private static void closeConnection(Connection connection) { } } + private static List> sortedByKey(Map documents) { + List> sorted = new ArrayList<>(documents.entrySet()); + sorted.sort(Comparator.comparing(entry -> entry.getKey().toString())); + return sorted; + } + private PreparedStatement getPreparedStatementForQuery( Map documents, String quotedPkColumn, @@ -1022,6 +1029,12 @@ private int executeBatchUpdate( List keys = keyGroup.getKeys(); List> allKeyParams = keyGroup.getKeyParams(); + Integer[] order = new Integer[keys.size()]; + for (int i = 0; i < order.length; i++) { + order[i] = i; + } + Arrays.sort(order, Comparator.comparing(i -> keys.get(i).toString())); + List setFragments = new ArrayList<>(keyGroup.getSetFragments()); List timestampParam = new ArrayList<>(); appendLastUpdatedTimestamp(setFragments, timestampParam, tableName, epochMillis); @@ -1034,7 +1047,7 @@ private int executeBatchUpdate( LOGGER.debug("Executing batch update SQL: {} for {} keys", sql, keys.size()); try (PreparedStatement ps = connection.prepareStatement(sql)) { - for (int i = 0; i < keys.size(); i++) { + for (int i : order) { int idx = 1; for (Object param : allKeyParams.get(i)) { ps.setObject(idx++, param); From 2819e9fa7e481a2ff0d919455e5f843da4b52d96 Mon Sep 17 00:00:00 2001 From: Prashant Pandey Date: Tue, 21 Jul 2026 14:50:50 +0530 Subject: [PATCH 2/3] Use TreeMap instead of LinkedHashMap to avoid explicit sorting --- .../postgres/FlatPostgresCollection.java | 23 +++++++++---------- 1 file changed, 11 insertions(+), 12 deletions(-) diff --git a/document-store/src/main/java/org/hypertrace/core/documentstore/postgres/FlatPostgresCollection.java b/document-store/src/main/java/org/hypertrace/core/documentstore/postgres/FlatPostgresCollection.java index 1c41cedb..f442d953 100644 --- a/document-store/src/main/java/org/hypertrace/core/documentstore/postgres/FlatPostgresCollection.java +++ b/document-store/src/main/java/org/hypertrace/core/documentstore/postgres/FlatPostgresCollection.java @@ -38,6 +38,7 @@ import java.util.Map.Entry; import java.util.Optional; import java.util.Set; +import java.util.TreeMap; import java.util.stream.Collectors; import org.hypertrace.core.documentstore.BulkArrayValueUpdateRequest; import org.hypertrace.core.documentstore.BulkDeleteResult; @@ -361,11 +362,13 @@ public boolean bulkUpsert(Map documents) { PostgresDataType pkType = getPrimaryKeyType(tableName, pkColumn); try { - // Parse all documents - Map parsedDocuments = new LinkedHashMap<>(); + // TreeMap keyed by Key.toString() so the downstream JDBC batch iterates rows in a canonical + // order. Postgres acquires row locks in batch-entry order, so overlapping concurrent batches + // must share an ordering to avoid deadlocks. + Map parsedDocuments = new TreeMap<>(Comparator.comparing(Key::toString)); List ignoredDocuments = new ArrayList<>(); - for (Map.Entry entry : sortedByKey(documents)) { + for (Map.Entry entry : documents.entrySet()) { List skippedFields = new ArrayList<>(); TypedDocument parsed = parseDocument(entry.getValue(), tableName, skippedFields); @@ -455,11 +458,13 @@ public boolean bulkCreateOrReplace(Map documents) { PostgresDataType pkType = getPrimaryKeyType(tableName, pkColumn); try { - // Parse all documents - Map parsedDocuments = new LinkedHashMap<>(); + // TreeMap keyed by Key.toString() so the downstream JDBC batch iterates rows in a canonical + // order. Postgres acquires row locks in batch-entry order, so overlapping concurrent batches + // must share an ordering to avoid deadlocks. + Map parsedDocuments = new TreeMap<>(Comparator.comparing(Key::toString)); List ignoredDocuments = new ArrayList<>(); - for (Map.Entry entry : sortedByKey(documents)) { + for (Map.Entry entry : documents.entrySet()) { List skippedFields = new ArrayList<>(); TypedDocument parsed = parseDocument(entry.getValue(), tableName, skippedFields); @@ -666,12 +671,6 @@ private static void closeConnection(Connection connection) { } } - private static List> sortedByKey(Map documents) { - List> sorted = new ArrayList<>(documents.entrySet()); - sorted.sort(Comparator.comparing(entry -> entry.getKey().toString())); - return sorted; - } - private PreparedStatement getPreparedStatementForQuery( Map documents, String quotedPkColumn, From 4b37f48812f657f8f24982edc7b18c8c1696d07d Mon Sep 17 00:00:00 2001 From: Prashant Pandey Date: Tue, 21 Jul 2026 14:52:04 +0530 Subject: [PATCH 3/3] Added comment --- .../core/documentstore/postgres/FlatPostgresCollection.java | 1 + 1 file changed, 1 insertion(+) diff --git a/document-store/src/main/java/org/hypertrace/core/documentstore/postgres/FlatPostgresCollection.java b/document-store/src/main/java/org/hypertrace/core/documentstore/postgres/FlatPostgresCollection.java index f442d953..2c8f263c 100644 --- a/document-store/src/main/java/org/hypertrace/core/documentstore/postgres/FlatPostgresCollection.java +++ b/document-store/src/main/java/org/hypertrace/core/documentstore/postgres/FlatPostgresCollection.java @@ -1032,6 +1032,7 @@ private int executeBatchUpdate( for (int i = 0; i < order.length; i++) { order[i] = i; } + // Sort keys to ensure consistent order to avoid deadlocks Arrays.sort(order, Comparator.comparing(i -> keys.get(i).toString())); List setFragments = new ArrayList<>(keyGroup.getSetFragments());