Switch back to use bulk values

OpenBMP · Mar 31, 2021 · a852c5c · a852c5c
1 parent f87fc6b
commit a852c5c
Show file tree

Hide file tree

Showing 6 changed files with 33 additions and 275 deletions.
diff --git a/src/main/java/org/openbmp/WriterRunnable.java b/src/main/java/org/openbmp/WriterRunnable.java
@@ -94,25 +94,32 @@ public void run() {
 
                         StringBuilder query = new StringBuilder();
 
-                        // Loop through queries and add them as multi-statements
+                        // Loop through queries and add as unique values to map, replacing duplicate (state compression)
                         for (Map.Entry<String, LinkedList<String>> entry : bulk_query.entrySet()) {
                             String key = entry.getKey().toString();
 
                             String[] ins = key.split("[|]");
+                            query.append(ins[0]);       // Insert statement
 
+                            boolean first = true;
                             for (String value : entry.getValue()) {
 
-                                if (query.length() > 0)
-                                    query.append(';');
+                                if (first) {
+                                    first = false;
+                                } else {
+                                    query.append(',');
+                                }
 
-                                query.append(ins[0]);
-                                query.append(' ');
                                 query.append(value);
-                                query.append(' ');
 
-                                if (ins.length > 1 && ins[1] != null && ins[1].length() > 0)
-                                    query.append(ins[1]);
                             }
+
+                            // Ending suffix statement, such as on conflict
+                            if (ins.length > 1 && ins[1] != null && ins[1].length() > 0)
+                                query.append(ins[1]);
+
+                            query.append(';');
+
                         }
 
                         if (query.length() > 0) {

diff --git a/src/main/java/org/openbmp/psqlquery/BaseAttributeQuery.java b/src/main/java/org/openbmp/psqlquery/BaseAttributeQuery.java
@@ -36,14 +36,13 @@ public String[] genInsertStatement() {
         final String [] stmt = { " INSERT INTO base_attrs (hash_id,peer_hash_id,origin,as_path,origin_as,next_hop,med,local_pref," +
                                  "isAtomicAgg,aggregator,community_list,ext_community_list,large_community_list," +
                                  "cluster_list,originator_id,as_path_count,nexthop_isIPv4,timestamp)" +
+//                                  " VALUES ",
+                                 "SELECT DISTINCT ON (hash_id) * FROM ( VALUES ",
 
-                                  " VALUES ",
-//                                 "SELECT DISTINCT ON (hash_id) * FROM ( VALUES ",
-//
-//                                 ") t(hash_id,peer_hash_id,origin,as_path,origin_as,next_hop,med,local_pref," +
-//                                      "isAtomicAgg,aggregator,community_list,ext_community_list,large_community_list," +
-//                                      "cluster_list,originator_id,as_path_count,nexthop_isIPv4,timestamp)" +
-//                                 " ORDER BY hash_id,timestamp desc" +
+                                 ") t(hash_id,peer_hash_id,origin,as_path,origin_as,next_hop,med,local_pref," +
+                                      "isAtomicAgg,aggregator,community_list,ext_community_list,large_community_list," +
+                                      "cluster_list,originator_id,as_path_count,nexthop_isIPv4,timestamp)" +
+                                 " ORDER BY hash_id,timestamp desc" +
                                     " ON CONFLICT (peer_hash_id,hash_id) DO UPDATE SET " +
                                             "timestamp=excluded.timestamp" };
         return stmt;

diff --git a/src/main/java/org/openbmp/psqlquery/UnicastPrefixQuery.java b/src/main/java/org/openbmp/psqlquery/UnicastPrefixQuery.java
@@ -40,13 +40,13 @@ public String[] genInsertStatement() {
                            "origin_as,prefix,prefix_len,prefix_bits,timestamp," +
                            "isWithdrawn,path_id,labels,isPrePolicy,isAdjRibIn) " +
 
-                            " VALUES ",
-//                            "SELECT DISTINCT ON (hash_id) * FROM ( VALUES ",
-//                            ") t(hash_id,peer_hash_id,base_attr_hash_id,isIPv4," +
-//                                "origin_as,prefix,prefix_len,prefix_bits,timestamp,"  +
-//                                "isWithdrawn,path_id,labels,isPrePolicy,isAdjRibIn) " +
-//                           " ORDER BY hash_id,timestamp desc" +
+//                            " VALUES ",
+                            "SELECT DISTINCT ON (hash_id) * FROM ( VALUES ",
 
+                            ") t(hash_id,peer_hash_id,base_attr_hash_id,isIPv4," +
+                                "origin_as,prefix,prefix_len,prefix_bits,timestamp,"  +
+                                "isWithdrawn,path_id,labels,isPrePolicy,isAdjRibIn) " +
+                           " ORDER BY hash_id,timestamp desc" +
                            " ON CONFLICT (peer_hash_id,hash_id) DO UPDATE SET timestamp=excluded.timestamp," +
                                "base_attr_hash_id=CASE excluded.isWithdrawn WHEN true THEN ip_rib.base_attr_hash_id ELSE excluded.base_attr_hash_id END," +
                                "origin_as=CASE excluded.isWithdrawn WHEN true THEN ip_rib.origin_as ELSE excluded.origin_as END," +

diff --git a/src/main/java/org/openbmp/psqlquery/oldLsLinkQuery.java b/src/main/java/org/openbmp/psqlquery/oldLsLinkQuery.java
diff --git a/src/main/java/org/openbmp/psqlquery/oldLsPrefixQuery.java b/src/main/java/org/openbmp/psqlquery/oldLsPrefixQuery.java
diff --git a/src/main/resources/obmp-psql.yml b/src/main/resources/obmp-psql.yml
@@ -40,15 +40,15 @@ base:
 
   # Number of seconds between rebalacing of writer threads
   #    Rebalance will drain writer queues at this interval if at least one writer is above threshold
-  writer_rebalance_seconds: 600
+  writer_rebalance_seconds: 480
 
   # Maximum input queue size
   #   Normally within the range of 1000 - 20000 is enough
-  writer_queue_size: 3000
+  writer_queue_size: 2500
 
   # Maximum input queue size for the consumer
   #   A good starting size is 2 times the size of the writer queue size
-  consumer_queue_size: 60000
+  consumer_queue_size: 30000
 
   # By default AS Path indexing is enabled.  This can be very resource intensive to psql
   #    and disk.   You can disable updating the psql 'as_path_analysis' table if
@@ -65,10 +65,11 @@ postgres:
 
   # The number of statements or records to batch in a single bulk update/insert/delete
   #   NOTE: It's more efficient to have more threads with a low batch size
-  batch_records: 300
+  batch_records: 500
 
   # The time in milliseconds to wait for batching records in a bulk update/insert/delete
-  batch_time_millis: 100
+  #  Note this will state compress records in this time period.
+  batch_time_millis: 200
 
   # The number of times to retry a statement
   retries: 10