ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [17/59] [abbrv] incubator-ignite git commit: # IGNITE-32: Fixes after review.
Date Thu, 05 Feb 2015 09:05:15 GMT
# IGNITE-32: Fixes after review.


Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/8e5c0a8a
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/8e5c0a8a
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/8e5c0a8a

Branch: refs/heads/ignite-103
Commit: 8e5c0a8af10e39dd4f70d9fcecb8c60d0b3d7822
Parents: 707aaf5
Author: AKuznetsov <akuznetsov@gridgain.com>
Authored: Tue Feb 3 09:26:03 2015 +0700
Committer: AKuznetsov <akuznetsov@gridgain.com>
Committed: Tue Feb 3 09:26:03 2015 +0700

----------------------------------------------------------------------
 .../ignite/cache/store/jdbc/JdbcCacheStore.java | 259 ++++++++++++-------
 1 file changed, 161 insertions(+), 98 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8e5c0a8a/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/JdbcCacheStore.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/JdbcCacheStore.java
b/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/JdbcCacheStore.java
index 4774d16..33461b7 100644
--- a/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/JdbcCacheStore.java
+++ b/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/JdbcCacheStore.java
@@ -169,8 +169,6 @@ public abstract class JdbcCacheStore<K, V> extends CacheStore<K,
V> implements L
     protected JdbcDialect resolveDialect() throws CacheException {
         Connection conn = null;
 
-        // TODO check conn.getMetaData().getURL() will work ???
-
         String dbProductName = null;
 
         try {
@@ -612,6 +610,44 @@ public abstract class JdbcCacheStore<K, V> extends CacheStore<K,
V> implements L
         }
     }
 
+    /**
+     * @param insStmt Insert statement.
+     * @param updStmt Update statement.
+     * @param em Entry mapping.
+     * @param entry Cache entry.
+     */
+    private void writeUpsert(PreparedStatement insStmt, PreparedStatement updStmt,
+        EntryMapping em, Cache.Entry<? extends K, ? extends V> entry) throws SQLException
{
+        for (int attempt = 0; attempt < MAX_ATTEMPT_WRITE_COUNT; attempt++) {
+            int i = fillValueParameters(updStmt, 1, em, entry.getValue());
+
+            fillKeyParameters(updStmt, i, em, entry.getKey());
+
+            if (updStmt.executeUpdate() == 0) {
+                i = fillKeyParameters(insStmt, em, entry.getKey());
+
+                fillValueParameters(insStmt, i, em, entry.getValue());
+
+                try {
+                    insStmt.executeUpdate();
+                }
+                catch (SQLException e) {
+                    // The error with code 23505 is thrown when trying to insert a row that
+                    // would violate a unique index or primary key.
+                    // TODO check with all RDBMS
+                    if (e.getErrorCode() == 23505)
+                        continue;
+
+                    throw e;
+                }
+            }
+
+            return;
+        }
+
+        throw new CacheWriterException("Failed write entry to database: " + entry);
+    }
+
     /** {@inheritDoc} */
     @Override public void write(Cache.Entry<? extends K, ? extends V> entry) throws
CacheWriterException {
         assert entry != null;
@@ -625,64 +661,49 @@ public abstract class JdbcCacheStore<K, V> extends CacheStore<K,
V> implements L
 
         Connection conn = null;
 
-        PreparedStatement stmt = null;
-
         try {
             conn = connection();
 
             if (dialect.hasMerge()) {
-                stmt = conn.prepareStatement(em.mergeQry);
+                PreparedStatement stmt = null;
+
+                try {
+                    stmt = conn.prepareStatement(em.mergeQry);
 
-                int i = fillKeyParameters(stmt, em, key);
+                    int i = fillKeyParameters(stmt, em, key);
 
-                fillValueParameters(stmt, i, em, entry.getValue());
+                    fillValueParameters(stmt, i, em, entry.getValue());
 
-                stmt.executeUpdate();
+                    stmt.executeUpdate();
+                }
+                finally {
+                    U.closeQuiet(stmt);
+                }
             }
             else {
-                V val = entry.getValue();
-
-                for (int attempt = 0; attempt < MAX_ATTEMPT_WRITE_COUNT; attempt++) {
-                    stmt = conn.prepareStatement(em.updQry);
+                PreparedStatement insStmt = null;
 
-                    int i = fillValueParameters(stmt, 1, em, val);
+                PreparedStatement updStmt = null;
 
-                    fillKeyParameters(stmt, i, em, key);
-
-                    if (stmt.executeUpdate() == 0) {
-                        U.closeQuiet(stmt);
-
-                        stmt = conn.prepareStatement(em.insQry);
-
-                        i = fillKeyParameters(stmt, em, key);
-
-                        fillValueParameters(stmt, i, em, val);
+                try {
+                    insStmt = conn.prepareStatement(em.insQry);
 
-                        try {
-                            stmt.executeUpdate();
-                        }
-                        catch (SQLException e) {
-                            // The error with code 23505 is thrown when trying to insert
a row that
-                            // would violate a unique index or primary key.
-                            // TODO check with all RDBMS
-                            if (e.getErrorCode() == 23505)
-                                continue;
-
-                            throw e;
-                        }
-                    }
+                    updStmt = conn.prepareStatement(em.updQry);
 
-                    return;
+                    writeUpsert(insStmt, updStmt, em, entry);
                 }
+                finally {
+                    U.closeQuiet(insStmt);
 
-                throw new CacheWriterException("Failed write entry to database: " + entry);
+                    U.closeQuiet(updStmt);
+                }
             }
         }
         catch (SQLException e) {
             throw new CacheWriterException("Failed to write entry to database: " + entry,
e);
         }
         finally {
-            end(conn, stmt);
+            closeConnection(conn);
         }
     }
 
@@ -691,68 +712,101 @@ public abstract class JdbcCacheStore<K, V> extends CacheStore<K,
V> implements L
         throws CacheWriterException {
         assert entries != null;
 
-        if (dialect.hasMerge()) {
-            Connection conn = null;
+        Connection conn = null;
 
-            PreparedStatement mergeStmt = null;
+        try {
+            conn = connection();
 
-            try {
-                conn = connection();
+            Object currKeyTypeId = null;
 
-                Object currKeyTypeId = null;
+            if (dialect.hasMerge()) {
+                PreparedStatement mergeStmt = null;
 
-                int cnt = 0;
+                try {
+                    int fromIdx = 0, prepared = 0;
 
-                for (Cache.Entry<? extends K, ? extends V> entry : entries) {
-                    K key = entry.getKey();
+                    for (Cache.Entry<? extends K, ? extends V> entry : entries) {
+                        K key = entry.getKey();
 
-                    Object keyTypeId = keyTypeId(key);
+                        Object keyTypeId = keyTypeId(key);
 
-                    EntryMapping em = entryMapping(keyTypeId, key);
+                        EntryMapping em = entryMapping(keyTypeId, key);
 
-                    if (mergeStmt == null) {
-                        mergeStmt = conn.prepareStatement(em.mergeQry);
+                        if (currKeyTypeId == null || !currKeyTypeId.equals(keyTypeId)) {
+                            if (mergeStmt != null) {
+                                executeBatch(mergeStmt, "writeAll", fromIdx, prepared, entries);
 
-                        currKeyTypeId = keyTypeId;
-                    }
+                                U.closeQuiet(mergeStmt);
+                            }
 
-                    if (!currKeyTypeId.equals(keyTypeId)) {
-                        executeBatch(mergeStmt, "writeAll", cnt);
+                            mergeStmt = conn.prepareStatement(em.mergeQry);
 
-                        currKeyTypeId = keyTypeId;
+                            currKeyTypeId = keyTypeId;
 
-                        cnt = 0;
-                    }
+                            prepared = 0;
+                        }
 
-                    int i = fillKeyParameters(mergeStmt, em, key);
+                        int i = fillKeyParameters(mergeStmt, em, key);
 
-                    fillValueParameters(mergeStmt, i, em, entry.getValue());
+                        fillValueParameters(mergeStmt, i, em, entry.getValue());
 
-                    mergeStmt.addBatch();
+                        mergeStmt.addBatch();
 
-                    if (++cnt % batchSz == 0) {
-                        executeBatch(mergeStmt, "writeAll", cnt);
+                        if (++prepared % batchSz == 0) {
+                            executeBatch(mergeStmt, "writeAll", fromIdx, prepared, entries);
 
-                        cnt = 0;
+                            prepared = 0;
+                        }
                     }
-                }
 
-                if (mergeStmt != null && cnt % batchSz != 0)
-                    executeBatch(mergeStmt, "writeAll", cnt);
-            }
-            catch (SQLException e) {
-                throw new CacheWriterException("Failed to write entries in database", e);
+                    if (mergeStmt != null && prepared % batchSz != 0)
+                        executeBatch(mergeStmt, "writeAll", fromIdx, prepared, entries);
+                }
+                finally {
+                    U.closeQuiet(mergeStmt);
+                }
             }
-            finally {
-                U.closeQuiet(mergeStmt);
+            else {
+                PreparedStatement insStmt = null;
+
+                PreparedStatement updStmt = null;
+
+                try {
+                    for (Cache.Entry<? extends K, ? extends V> entry : entries) {
+                        K key = entry.getKey();
+
+                        Object keyTypeId = keyTypeId(key);
+
+                        EntryMapping em = entryMapping(keyTypeId, key);
 
-                closeConnection(conn);
+                        if (currKeyTypeId == null || !currKeyTypeId.equals(keyTypeId)) {
+                            U.closeQuiet(insStmt);
+
+                            insStmt = conn.prepareStatement(em.insQry);
+
+                            U.closeQuiet(updStmt);
+
+                            updStmt = conn.prepareStatement(em.updQry);
+
+                            currKeyTypeId = keyTypeId;
+                        }
+
+                        writeUpsert(insStmt, updStmt, em, entry);
+                    }
+                }
+                finally {
+                    U.closeQuiet(insStmt);
+
+                    U.closeQuiet(updStmt);
+                }
             }
         }
-        else
-            for (Cache.Entry<? extends K, ? extends V> e : entries)
-                write(e); // TODO rework to optimal usage. Method write will get all params
each time (conn, stmt).
-                          // split into 2 methods writeMerge + writeUpsert.
+        catch (SQLException e) {
+            throw new CacheWriterException("Failed to write entries in database", e);
+        }
+        finally {
+            closeConnection(conn);
+        }
     }
 
     /** {@inheritDoc} */
@@ -788,24 +842,29 @@ public abstract class JdbcCacheStore<K, V> extends CacheStore<K,
V> implements L
 
     /**
      * @param stmt Statement.
-     * @param stmtType Statement type for error message.
-     * @param batchSz Expected batch size.
+     * @param stmtType Statement description for error message.
+     * @param fromIdx Objects in batch start from index.
+     * @param prepared Expected objects in batch.
+     * @param objects All objects.
      */
-    private void executeBatch(Statement stmt, String stmtType, int batchSz) throws SQLException
{
+    private void executeBatch(Statement stmt, String stmtType, int fromIdx, int prepared,
Collection<?> objects)
+        throws SQLException {
         int[] rowCounts = stmt.executeBatch();
 
         int numOfRowCnt = rowCounts.length;
 
-        if (numOfRowCnt != batchSz)
-            log.warning("JDBC driver did not return the expected number of row counts," +
-                " actual row count: " + numOfRowCnt + " expected: " + batchSz);
+        if (numOfRowCnt != prepared)
+            log.warning("JDBC driver did not return the expected number of updated row counts,"
+
+                " actual row count: " + numOfRowCnt + " expected: " + prepared);
+
+        Object[] arr = null;
+
+        for (int i = 0; i < numOfRowCnt; i++)
+            if (rowCounts[i] != 1) {
+                if (arr == null)
+                    arr = objects.toArray();
 
-        for (int rowCount : rowCounts)
-            if (rowCount != 1) {
-                // TODO stmtType used 2 times?
-                // TODO print warning for all keys.
-                // TODO while failed - convert collection to array and print warnins to each
failed key.
-                log.warning("Batch " + stmtType + " returned unexpected row count from "
+ stmtType + " statement");
+                log.warning("Batch " + stmtType + " returned unexpected updated row count
for: " + arr[fromIdx + i]);
             }
     }
 
@@ -822,7 +881,7 @@ public abstract class JdbcCacheStore<K, V> extends CacheStore<K,
V> implements L
 
             PreparedStatement delStmt = null;
 
-            int cnt = 0;
+            int fromIdx = 0, prepared = 0;
 
             for (Object key : keys) {
                 Object keyTypeId = keyTypeId(key);
@@ -836,9 +895,11 @@ public abstract class JdbcCacheStore<K, V> extends CacheStore<K,
V> implements L
                 }
 
                 if (!currKeyTypeId.equals(keyTypeId)) {
-                    executeBatch(delStmt, "deleteAll", cnt);
+                    executeBatch(delStmt, "deleteAll", fromIdx, prepared, keys);
 
-                    cnt = 0;
+                    fromIdx += prepared;
+
+                    prepared = 0;
 
                     currKeyTypeId = keyTypeId;
                 }
@@ -847,15 +908,17 @@ public abstract class JdbcCacheStore<K, V> extends CacheStore<K,
V> implements L
 
                 delStmt.addBatch();
 
-                if (++cnt % batchSz == 0) {
-                    executeBatch(delStmt, "deleteAll", cnt);
+                if (++prepared % batchSz == 0) {
+                    executeBatch(delStmt, "deleteAll", fromIdx, prepared, keys);
+
+                    fromIdx += prepared;
 
-                    cnt = 0;
+                    prepared = 0;
                 }
             }
 
-            if (delStmt != null && cnt % batchSz != 0)
-                executeBatch(delStmt, "deleteAll", cnt);
+            if (delStmt != null && prepared % batchSz != 0)
+                executeBatch(delStmt, "deleteAll", fromIdx, prepared, keys);
         }
         catch (SQLException e) {
             throw new CacheWriterException("Failed to remove values from database", e);


Mime
View raw message