From 0e229132f1178a50db67cd795eb7e7f64fb214b2 Mon Sep 17 00:00:00 2001
From: erilong <erilong@users.noreply.github.com>
Date: Thu, 6 Mar 2008 14:25:02 +0000
Subject: [PATCH] [1908764] Incoming batch timings for network, filter, and
 database access

---
 .../symmetric/load/DataLoaderStatistics.java  | 66 +++++++++++++++++++
 .../symmetric/load/IDataLoaderStatistics.java |  8 +++
 .../symmetric/load/csv/CsvLoader.java         | 18 ++++-
 .../symmetric/model/IncomingBatchHistory.java | 44 +++++++++++++
 .../service/impl/IncomingBatchService.java    | 17 +++--
 symmetric/src/main/resources/ddl-config.xml   |  4 ++
 .../src/main/resources/symmetric-services.xml |  6 +-
 .../src/main/resources/symmetric-upgrade.xml  | 12 ++++
 .../service/impl/DataLoaderServiceTest.java   |  1 +
 9 files changed, 167 insertions(+), 9 deletions(-)

diff --git a/symmetric/src/main/java/org/jumpmind/symmetric/load/DataLoaderStatistics.java b/symmetric/src/main/java/org/jumpmind/symmetric/load/DataLoaderStatistics.java
index 8c2c8186e9..df54c77be3 100644
--- a/symmetric/src/main/java/org/jumpmind/symmetric/load/DataLoaderStatistics.java
+++ b/symmetric/src/main/java/org/jumpmind/symmetric/load/DataLoaderStatistics.java
@@ -27,6 +27,14 @@ public class DataLoaderStatistics implements IDataLoaderStatistics {
 
     private Date startTime;
     
+    private long networkMillis;
+    
+    private long filterMillis;
+
+    private long databaseMillis;
+    
+    private long byteCount;
+    
     private long lineCount;
     
     private long statementCount;
@@ -36,6 +44,8 @@ public class DataLoaderStatistics implements IDataLoaderStatistics {
     private long fallbackUpdateCount;
     
     private long missingDeleteCount;
+    
+    private long timerMillis;
 
     public DataLoaderStatistics() {
         this.startTime = new Date();
@@ -60,7 +70,31 @@ public long incrementMissingDeleteCount() {
     public long incrementStatementCount() {
         return ++statementCount;
     }
+    
+    public void incrementNetworkMillis(long millis) {
+        networkMillis += millis;
+    }
+
+    public void incrementFilterMillis(long millis) {
+        filterMillis += millis;
+    }
 
+    public void incrementDatabaseMillis(long millis) {
+        databaseMillis += millis;
+    }
+
+    public void incrementByteCount(long count) {
+        byteCount += count;
+    }
+
+    public void startTimer() {
+        timerMillis = System.currentTimeMillis();
+    }
+    
+    public long endTimer() {
+        return System.currentTimeMillis() - timerMillis;
+    }
+    
     public long getFallbackInsertCount() {
         return fallbackInsertCount;
     }
@@ -109,4 +143,36 @@ public void setMissingDeleteCount(long missingDeleteCount) {
         this.missingDeleteCount = missingDeleteCount;
     }
 
+    public long getDatabaseMillis() {
+        return databaseMillis;
+    }
+
+    public void setDatabaseMillis(long databaseMillis) {
+        this.databaseMillis = databaseMillis;
+    }
+
+    public long getFilterMillis() {
+        return filterMillis;
+    }
+
+    public void setFilterMillis(long filterMillis) {
+        this.filterMillis = filterMillis;
+    }
+
+    public long getNetworkMillis() {
+        return networkMillis;
+    }
+
+    public void setNetworkMillis(long networkMillis) {
+        this.networkMillis = networkMillis;
+    }
+
+    public long getByteCount() {
+        return byteCount;
+    }
+
+    public void setByteCount(long byteCount) {
+        this.byteCount = byteCount;
+    }
+
 }
diff --git a/symmetric/src/main/java/org/jumpmind/symmetric/load/IDataLoaderStatistics.java b/symmetric/src/main/java/org/jumpmind/symmetric/load/IDataLoaderStatistics.java
index 6df0ed724b..5637f2532f 100644
--- a/symmetric/src/main/java/org/jumpmind/symmetric/load/IDataLoaderStatistics.java
+++ b/symmetric/src/main/java/org/jumpmind/symmetric/load/IDataLoaderStatistics.java
@@ -25,6 +25,14 @@
 
 public interface IDataLoaderStatistics {
 
+    public long getByteCount();
+    
+    public long getNetworkMillis();
+    
+    public long getFilterMillis();
+    
+    public long getDatabaseMillis();
+    
     public long getFallbackInsertCount();
 
     public long getFallbackUpdateCount();
diff --git a/symmetric/src/main/java/org/jumpmind/symmetric/load/csv/CsvLoader.java b/symmetric/src/main/java/org/jumpmind/symmetric/load/csv/CsvLoader.java
index a6e2f4216d..8147f5aaf7 100644
--- a/symmetric/src/main/java/org/jumpmind/symmetric/load/csv/CsvLoader.java
+++ b/symmetric/src/main/java/org/jumpmind/symmetric/load/csv/CsvLoader.java
@@ -87,6 +87,7 @@ public void open(BufferedReader reader, List<IDataLoaderFilter> filters, Map<Str
     public boolean hasNext() throws IOException {
         while (csvReader.readRecord()) {
             String[] tokens = csvReader.getValues();
+            stats.incrementByteCount(csvReader.getRawRecord().length());
 
             if (tokens[0].equals(CsvConstants.BATCH)) {
                 context.setBatchId(tokens[1]);
@@ -116,6 +117,7 @@ public void load() throws IOException {
         while (csvReader.readRecord()) {
             String[] tokens = csvReader.getValues();
             stats.incrementLineCount();
+            stats.incrementByteCount(csvReader.getRawRecord().length());
 
             if (tokens[0].equals(CsvConstants.INSERT)) {
                 if (!context.getTableTemplate().isIgnoreThisTable() && !context.isSkipping()) {
@@ -192,13 +194,16 @@ protected int insert(String[] tokens, BinaryEncoding encoding) {
         int rows = 0;
 
         if (filters != null) {
+            stats.startTimer();
             for (IDataLoaderFilter filter : filters) {
                 filter.filterInsert(context, columnValues);
             }
+            stats.incrementFilterMillis(stats.endTimer());
         }
 
         Object savepoint = null;
         try {
+            stats.startTimer();
             if (enableFallbackUpdate) {
                 savepoint = dbDialect.createSavepointForFallback();
             }
@@ -223,6 +228,8 @@ protected int insert(String[] tokens, BinaryEncoding encoding) {
                 // TODO: log the PK information as an ERROR level.
                 throw e;
             }
+        } finally {
+            stats.incrementDatabaseMillis(stats.endTimer());
         }
         return rows;
     }
@@ -233,11 +240,14 @@ protected int update(String[] tokens, BinaryEncoding encoding) {
         String keyValues[] = parseKeys(tokens, 1 + columnValues.length);
 
         if (filters != null) {
+            stats.startTimer();
             for (IDataLoaderFilter filter : filters) {
                 filter.filterUpdate(context, columnValues, keyValues);
             }
+            stats.incrementFilterMillis(stats.endTimer());
         }
 
+        stats.startTimer();
         int rows = context.getTableTemplate().update(columnValues, keyValues, encoding);
         if (rows == 0) {
             if (enableFallbackInsert) {
@@ -246,9 +256,10 @@ protected int update(String[] tokens, BinaryEncoding encoding) {
                             + ArrayUtils.toString(tokens));
                 }
                 stats.incrementFallbackInsertCount();
-                return context.getTableTemplate().insert(columnValues, encoding);
+                rows = context.getTableTemplate().insert(columnValues, encoding);
             } else {
                 // TODO: log the PK information as an ERROR level.
+                stats.incrementDatabaseMillis(stats.endTimer());
                 throw new RuntimeException("Unable to update " + context.getTableName() + ": "
                         + ArrayUtils.toString(tokens));
             }
@@ -256,6 +267,7 @@ protected int update(String[] tokens, BinaryEncoding encoding) {
             logger.warn("Too many rows (" + rows + ") updated for " + context.getTableName() + ": "
                     + ArrayUtils.toString(tokens));
         }
+        stats.incrementDatabaseMillis(stats.endTimer());
         return rows;
     }
 
@@ -264,12 +276,16 @@ protected int delete(String[] tokens) {
         String keyValues[] = parseKeys(tokens, 1);
 
         if (filters != null) {
+            stats.startTimer();
             for (IDataLoaderFilter filter : filters) {
                 filter.filterDelete(context, keyValues);
             }
+            stats.incrementFilterMillis(stats.endTimer());
         }
 
+        stats.startTimer();
         int rows = context.getTableTemplate().delete(keyValues);
+        stats.incrementDatabaseMillis(stats.endTimer());
         if (rows == 0) {
             if (allowMissingDelete) {
                 logger
diff --git a/symmetric/src/main/java/org/jumpmind/symmetric/model/IncomingBatchHistory.java b/symmetric/src/main/java/org/jumpmind/symmetric/model/IncomingBatchHistory.java
index 60d8fe7be8..627a8cfead 100644
--- a/symmetric/src/main/java/org/jumpmind/symmetric/model/IncomingBatchHistory.java
+++ b/symmetric/src/main/java/org/jumpmind/symmetric/model/IncomingBatchHistory.java
@@ -45,6 +45,14 @@ public enum Status {
 
     private static String hostName;
 
+    private long byteCount;
+
+    private long networkMillis;
+    
+    private long filterMillis;
+
+    private long databaseMillis;
+
     private long statementCount;
 
     private long fallbackInsertCount;
@@ -80,6 +88,10 @@ public IncomingBatchHistory(IDataLoaderContext context) {
     }
     
     public void setValues(IDataLoaderStatistics statistics, boolean isSuccess) {
+        byteCount = statistics.getByteCount();
+        networkMillis = statistics.getNetworkMillis();
+        filterMillis = statistics.getFilterMillis();
+        databaseMillis = statistics.getDatabaseMillis();
         statementCount = statistics.getStatementCount();
         fallbackInsertCount = statistics.getFallbackInsertCount();
         fallbackUpdateCount = statistics.getFallbackUpdateCount();
@@ -179,4 +191,36 @@ public void setMissingDeleteCount(long missingDeleteCount) {
         this.missingDeleteCount = missingDeleteCount;
     }
 
+    public long getByteCount() {
+        return byteCount;
+    }
+
+    public void setByteCount(long byteCount) {
+        this.byteCount = byteCount;
+    }
+
+    public long getDatabaseMillis() {
+        return databaseMillis;
+    }
+
+    public void setDatabaseMillis(long databaseMillis) {
+        this.databaseMillis = databaseMillis;
+    }
+
+    public long getFilterMillis() {
+        return filterMillis;
+    }
+
+    public void setFilterMillis(long filterMillis) {
+        this.filterMillis = filterMillis;
+    }
+
+    public long getNetworkMillis() {
+        return networkMillis;
+    }
+
+    public void setNetworkMillis(long networkMillis) {
+        this.networkMillis = networkMillis;
+    }
+
 }
diff --git a/symmetric/src/main/java/org/jumpmind/symmetric/service/impl/IncomingBatchService.java b/symmetric/src/main/java/org/jumpmind/symmetric/service/impl/IncomingBatchService.java
index 0315bdb012..69fee5a29a 100644
--- a/symmetric/src/main/java/org/jumpmind/symmetric/service/impl/IncomingBatchService.java
+++ b/symmetric/src/main/java/org/jumpmind/symmetric/service/impl/IncomingBatchService.java
@@ -108,8 +108,9 @@ public int updateIncomingBatch(IncomingBatch status) {
 
     public void insertIncomingBatchHistory(IncomingBatchHistory history) {
         jdbcTemplate.update(insertIncomingBatchHistorySql, new Object[] { Long.valueOf(history.getBatchId()),
-                history.getNodeId(), history.getStatus().toString(), history.getHostName(),
-                history.getStatementCount(), history.getFallbackInsertCount(),
+                history.getNodeId(), history.getStatus().toString(), history.getNetworkMillis(),
+                history.getFilterMillis(), history.getDatabaseMillis(), history.getHostName(),
+                history.getByteCount(), history.getStatementCount(), history.getFallbackInsertCount(),
                 history.getFallbackUpdateCount(), history.getMissingDeleteCount(),
                 history.getFailedRowNumber(), history.getStartTime(), history.getEndTime() });
     }
@@ -134,10 +135,14 @@ public Object mapRow(ResultSet rs, int num) throws SQLException {
             history.setStartTime(rs.getTime(4));
             history.setEndTime(rs.getTime(5));
             history.setFailedRowNumber(rs.getLong(6));
-            history.setStatementCount(rs.getLong(7));
-            history.setFallbackInsertCount(rs.getLong(8));
-            history.setFallbackUpdateCount(rs.getLong(9));
-            history.setMissingDeleteCount(rs.getLong(10));
+            history.setByteCount(rs.getLong(7));
+            history.setNetworkMillis(rs.getLong(8));
+            history.setFilterMillis(rs.getLong(9));
+            history.setDatabaseMillis(rs.getLong(10));
+            history.setStatementCount(rs.getLong(11));
+            history.setFallbackInsertCount(rs.getLong(12));
+            history.setFallbackUpdateCount(rs.getLong(13));
+            history.setMissingDeleteCount(rs.getLong(14));
             return history;
         }
     }
diff --git a/symmetric/src/main/resources/ddl-config.xml b/symmetric/src/main/resources/ddl-config.xml
index 8520125f51..7ee724de4b 100644
--- a/symmetric/src/main/resources/ddl-config.xml
+++ b/symmetric/src/main/resources/ddl-config.xml
@@ -241,8 +241,12 @@
         <column name="status" type="CHAR" size="2" />
         <column name="start_time" type="TIMESTAMP" />
         <column name="end_time" type="TIMESTAMP" />
+        <column name="network_millis" type="INTEGER" />
+        <column name="filter_millis" type="INTEGER" />
+        <column name="database_millis" type="INTEGER" />
         <column name="host_name" type="VARCHAR" size="50" />
         <column name="failed_row_number" type="INTEGER" />
+        <column name="byte_count" type="INTEGER" />
         <column name="statement_count" type="INTEGER" />
         <column name="fallback_insert_count" type="INTEGER" />
         <column name="fallback_update_count" type="INTEGER" />
diff --git a/symmetric/src/main/resources/symmetric-services.xml b/symmetric/src/main/resources/symmetric-services.xml
index 671e3e12f7..33e518fdc7 100644
--- a/symmetric/src/main/resources/symmetric-services.xml
+++ b/symmetric/src/main/resources/symmetric-services.xml
@@ -401,6 +401,7 @@
         <property name="findIncomingBatchHistorySql">
             <value>
                 select batch_id, node_id, status, start_time, end_time, failed_row_number,
+                byte_count, network_millis, filter_millis, database_millis,
                 statement_count, fallback_insert_count, fallback_update_count, missing_delete_count
                 from ${sync.table.prefix}_incoming_batch_hist where batch_id = ? and node_id = ?
                 order by start_time
@@ -421,9 +422,10 @@
         <property name="insertIncomingBatchHistorySql">
             <value>
                 insert into ${sync.table.prefix}_incoming_batch_hist (batch_id, node_id, status,
-                host_name, statement_count, fallback_insert_count, fallback_update_count,
+                network_millis, filter_millis, database_millis,
+                host_name, byte_count, statement_count, fallback_insert_count, fallback_update_count,
                 missing_delete_count, failed_row_number, start_time, end_time) values (?, ?, ?, ?,
-                ?, ?, ?, ?, ?, ?, ?)
+                ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
             </value>
         </property>
     </bean>
diff --git a/symmetric/src/main/resources/symmetric-upgrade.xml b/symmetric/src/main/resources/symmetric-upgrade.xml
index dc1fd309d2..1e75919ccb 100644
--- a/symmetric/src/main/resources/symmetric-upgrade.xml
+++ b/symmetric/src/main/resources/symmetric-upgrade.xml
@@ -183,6 +183,18 @@
                   create index idx_de_btchd_nid_cid on 
                     ${sync.table.prefix}_data_event(batched, node_id, channel_id)
                 </value>
+                <value>
+                  alter table ${sync.table.prefix}_incoming_batch_hist add (byte_count numeric(22))
+                </value>                
+                <value>
+                  alter table ${sync.table.prefix}_incoming_batch_hist add (network_millis numeric(22))
+                </value>                
+                <value>
+                  alter table ${sync.table.prefix}_incoming_batch_hist add (filter_millis numeric(22))
+                </value>                
+                <value>
+                  alter table ${sync.table.prefix}_incoming_batch_hist add (database_millis numeric(22))
+                </value>                
             </list>
         </property>
     </bean>
diff --git a/symmetric/src/test/java/org/jumpmind/symmetric/service/impl/DataLoaderServiceTest.java b/symmetric/src/test/java/org/jumpmind/symmetric/service/impl/DataLoaderServiceTest.java
index e6f98063df..b4d7e6e1f9 100644
--- a/symmetric/src/test/java/org/jumpmind/symmetric/service/impl/DataLoaderServiceTest.java
+++ b/symmetric/src/test/java/org/jumpmind/symmetric/service/impl/DataLoaderServiceTest.java
@@ -126,6 +126,7 @@ public void testStatistics() throws Exception {
         Assert.assertNotNull(history.getStartTime(), "Start time cannot be null");
         Assert.assertNotNull(history.getEndTime(), "End time cannot be null");
         Assert.assertEquals(history.getFailedRowNumber(), 8, "Wrong failed row number");
+        Assert.assertEquals(history.getByteCount(), 290, "Wrong byte count");
         Assert.assertEquals(history.getStatementCount(), 8, "Wrong statement count");
         Assert.assertEquals(history.getFallbackInsertCount(), 1, "Wrong fallback insert count");
         Assert.assertEquals(history.getFallbackUpdateCount(), 2, "Wrong fallback update count");