ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From akuznet...@apache.org
Subject ignite git commit: IGNITE-1753 WIP on POJO store refactoring + fixed first POJO store test.
Date Fri, 23 Oct 2015 16:24:41 GMT
Repository: ignite
Updated Branches:
  refs/heads/ignite-1753 aee462759 -> 8e26c84d1


IGNITE-1753 WIP on POJO store refactoring + fixed first POJO store test.


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

Branch: refs/heads/ignite-1753
Commit: 8e26c84d1a5739ed97c90c28e65bd075fcab1ade
Parents: aee4627
Author: AKuznetsov <akuznetsov@gridgain.com>
Authored: Fri Oct 23 23:24:32 2015 +0700
Committer: AKuznetsov <akuznetsov@gridgain.com>
Committed: Fri Oct 23 23:24:32 2015 +0700

----------------------------------------------------------------------
 .../cache/store/jdbc/CacheJdbcPojoStore.java    |  30 +-
 .../jdbc/CacheJdbcPojoStoreConfiguration.java   |   2 +-
 .../store/jdbc/CacheJdbcPojoStoreFactory.java   |   1 +
 .../store/jdbc/CacheJdbcPojoStoreType.java      |  14 +-
 .../CacheJdbcPojoStoreCompatibilityTest.java    | 407 +++++++++++++++++++
 .../store/jdbc/CacheJdbcPojoStoreTest.java      | 162 +++++---
 6 files changed, 527 insertions(+), 89 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/ignite/blob/8e26c84d/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStore.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStore.java
b/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStore.java
index 044f79d..a238f6e 100644
--- a/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStore.java
+++ b/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStore.java
@@ -246,7 +246,7 @@ public class CacheJdbcPojoStore<K, V> implements CacheStore<K,
V>, LifecycleAwar
      * @return Key type id.
      * @throws CacheException If failed to get type key id from object.
      */
-    protected Object keyTypeId(String type) throws CacheException {
+    private Object keyTypeId(String type) throws CacheException {
         try {
             return Class.forName(type);
         }
@@ -262,7 +262,7 @@ public class CacheJdbcPojoStore<K, V> implements CacheStore<K,
V>, LifecycleAwar
      * @param types Collection of types.
      * @throws CacheException If failed to prepare internal builders for types.
      */
-    protected void prepareBuilders(@Nullable String cacheName, Collection<CacheJdbcPojoStoreType>
types)
+    private void prepareBuilders(@Nullable String cacheName, Collection<CacheJdbcPojoStoreType>
types)
         throws CacheException {
         Map<String, PojoMethodsCache> typeMethods = U.newHashMap(types.size() * 2);
 
@@ -287,7 +287,7 @@ public class CacheJdbcPojoStore<K, V> implements CacheStore<K,
V>, LifecycleAwar
      * @return The resolved dialect.
      * @throws CacheException Indicates problems accessing the metadata.
      */
-    protected JdbcDialect resolveDialect() throws CacheException {
+    private JdbcDialect resolveDialect() throws CacheException {
         Connection conn = null;
 
         String dbProductName = null;
@@ -361,7 +361,7 @@ public class CacheJdbcPojoStore<K, V> implements CacheStore<K,
V>, LifecycleAwar
      * @return Connection.
      * @throws SQLException In case of error.
      */
-    protected Connection connection() throws SQLException {
+    private Connection connection() throws SQLException {
         CacheStoreSession ses = session();
 
         if (ses.transaction() != null) {
@@ -388,7 +388,7 @@ public class CacheJdbcPojoStore<K, V> implements CacheStore<K,
V>, LifecycleAwar
      *
      * @param conn Connection to close.
      */
-    protected void closeConnection(@Nullable Connection conn) {
+    private void closeConnection(@Nullable Connection conn) {
         CacheStoreSession ses = session();
 
         // Close connection right away if there is no transaction.
@@ -402,7 +402,7 @@ public class CacheJdbcPojoStore<K, V> implements CacheStore<K,
V>, LifecycleAwar
      * @param conn Allocated connection.
      * @param st Created statement,
      */
-    protected void end(@Nullable Connection conn, @Nullable Statement st) {
+    private void end(@Nullable Connection conn, @Nullable Statement st) {
         U.closeQuiet(st);
 
         closeConnection(conn);
@@ -448,7 +448,7 @@ public class CacheJdbcPojoStore<K, V> implements CacheStore<K,
V>, LifecycleAwar
      * @return Value in column.
      * @throws SQLException If a database access error occurs or this method is called.
      */
-    protected Object getColumnValue(ResultSet rs, int colIdx, Class<?> type) throws
SQLException {
+    private Object getColumnValue(ResultSet rs, int colIdx, Class<?> type) throws SQLException
{
         Object val = rs.getObject(colIdx);
 
         if (val == null)
@@ -589,7 +589,7 @@ public class CacheJdbcPojoStore<K, V> implements CacheStore<K,
V>, LifecycleAwar
      * @param cls Class.
      * @return {@code True} if object is a simple type.
      */
-    protected static boolean simpleType(Class<?> cls) {
+    private static boolean simpleType(Class<?> cls) {
         return (Number.class.isAssignableFrom(cls) || String.class.isAssignableFrom(cls)
||
             java.util.Date.class.isAssignableFrom(cls) || Boolean.class.isAssignableFrom(cls)
||
             UUID.class.isAssignableFrom(cls));
@@ -1394,7 +1394,7 @@ public class CacheJdbcPojoStore<K, V> implements CacheStore<K,
V>, LifecycleAwar
      * @param fieldVal Field value.
      * @throws CacheException If failed to set statement parameter.
      */
-    protected void fillParameter(PreparedStatement stmt, int i, CacheJdbcPojoStoreTypeField
field, @Nullable Object fieldVal)
+    private void fillParameter(PreparedStatement stmt, int i, CacheJdbcPojoStoreTypeField
field, @Nullable Object fieldVal)
         throws CacheException {
         try {
             if (fieldVal != null) {
@@ -1430,7 +1430,7 @@ public class CacheJdbcPojoStore<K, V> implements CacheStore<K,
V>, LifecycleAwar
      * @return Next index for parameters.
      * @throws CacheException If failed to set statement parameters.
      */
-    protected int fillKeyParameters(PreparedStatement stmt, int idx, EntryMapping em,
+    private int fillKeyParameters(PreparedStatement stmt, int idx, EntryMapping em,
         Object key) throws CacheException {
         for (CacheJdbcPojoStoreTypeField field : em.keyColumns()) {
             Object fieldVal = extractParameter(em.cacheName, em.keyType(), field.getJavaName(),
key);
@@ -1448,7 +1448,7 @@ public class CacheJdbcPojoStore<K, V> implements CacheStore<K,
V>, LifecycleAwar
      * @return Next index for parameters.
      * @throws CacheException If failed to set statement parameters.
      */
-    protected int fillKeyParameters(PreparedStatement stmt, EntryMapping m, Object key) throws
CacheException {
+    private int fillKeyParameters(PreparedStatement stmt, EntryMapping m, Object key) throws
CacheException {
         return fillKeyParameters(stmt, 1, m, key);
     }
 
@@ -1460,7 +1460,7 @@ public class CacheJdbcPojoStore<K, V> implements CacheStore<K,
V>, LifecycleAwar
      * @return Next index for parameters.
      * @throws CacheException If failed to set statement parameters.
      */
-    protected int fillValueParameters(PreparedStatement stmt, int idx, EntryMapping em, Object
val)
+    private int fillValueParameters(PreparedStatement stmt, int idx, EntryMapping em, Object
val)
         throws CacheWriterException {
         for (CacheJdbcPojoStoreTypeField field : em.uniqValFields) {
             Object fieldVal = extractParameter(em.cacheName, em.valueType(), field.getJavaName(),
val);
@@ -1618,7 +1618,7 @@ public class CacheJdbcPojoStore<K, V> implements CacheStore<K,
V>, LifecycleAwar
      */
     private static class PojoMethodsCache {
         /** POJO class. */
-        protected final Class<?> cls;
+        private final Class<?> cls;
 
         /** Constructor for POJO object. */
         private Constructor ctor;
@@ -1836,7 +1836,7 @@ public class CacheJdbcPojoStore<K, V> implements CacheStore<K,
V>, LifecycleAwar
          * @param keyCnt Key count.
          * @return Load query statement text.
          */
-        protected String loadQuery(int keyCnt) {
+        private String loadQuery(int keyCnt) {
             assert keyCnt <= maxKeysPerStmt;
 
             if (keyCnt == maxKeysPerStmt)
@@ -1855,7 +1855,7 @@ public class CacheJdbcPojoStore<K, V> implements CacheStore<K,
V>, LifecycleAwar
          * @param appendUpperBound Need add upper bound for range.
          * @return Query with range.
          */
-        protected String loadCacheRangeQuery(boolean appendLowerBound, boolean appendUpperBound)
{
+        private String loadCacheRangeQuery(boolean appendLowerBound, boolean appendUpperBound)
{
             return dialect.loadCacheRangeQuery(fullTblName, keyCols, cols, appendLowerBound,
appendUpperBound);
         }
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/8e26c84d/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStoreConfiguration.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStoreConfiguration.java
b/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStoreConfiguration.java
index 7322279..b333bc7 100644
--- a/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStoreConfiguration.java
+++ b/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStoreConfiguration.java
@@ -44,7 +44,7 @@ public class CacheJdbcPojoStoreConfiguration implements Serializable {
     private String dataSrcBean;
 
     /** Database dialect. */
-    protected JdbcDialect dialect;
+    private JdbcDialect dialect;
 
     /** Max workers thread count. These threads are responsible for load cache. */
     private int maxPoolSz = Runtime.getRuntime().availableProcessors();

http://git-wip-us.apache.org/repos/asf/ignite/blob/8e26c84d/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStoreFactory.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStoreFactory.java
b/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStoreFactory.java
index 003cfdb..6d8f8af 100644
--- a/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStoreFactory.java
+++ b/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStoreFactory.java
@@ -145,6 +145,7 @@ public class CacheJdbcPojoStoreFactory<K, V> implements Factory<CacheJdbcPojoSto
      * @return {@code This} for chaining.
      * @see CacheJdbcPojoStore#setDataSource(DataSource)
      */
+    @Deprecated
     public CacheJdbcPojoStoreFactory<K, V> setDataSourceBean(String dataSrcBean) {
         this.dataSrcBean = dataSrcBean;
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/8e26c84d/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStoreType.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStoreType.java
b/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStoreType.java
index 85d5ed5..e755165 100644
--- a/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStoreType.java
+++ b/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStoreType.java
@@ -81,21 +81,21 @@ public class CacheJdbcPojoStoreType implements Serializable {
     }
 
     /**
-     * Gets database schema name.
+     * Gets associated cache name.
      *
-     * @return Schema name.
+     * @return –°ache name.
      */
     public String getCacheName() {
-        return dbSchema;
+        return cacheName;
     }
 
     /**
-     * Sets database schema name.
+     * Sets associated cache name.
      *
-     * @param dbSchema Schema name.
+     * @param cacheName Cache name.
      */
-    public CacheJdbcPojoStoreType setCacheName(String dbSchema) {
-        this.dbSchema = dbSchema;
+    public CacheJdbcPojoStoreType setCacheName(String cacheName) {
+        this.cacheName = cacheName;
 
         return this;
     }

http://git-wip-us.apache.org/repos/asf/ignite/blob/8e26c84d/modules/core/src/test/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStoreCompatibilityTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStoreCompatibilityTest.java
b/modules/core/src/test/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStoreCompatibilityTest.java
new file mode 100644
index 0000000..021a89d
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStoreCompatibilityTest.java
@@ -0,0 +1,407 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.cache.store.jdbc;
+
+import org.apache.ignite.*;
+import org.apache.ignite.cache.store.jdbc.dialect.*;
+import org.apache.ignite.cache.store.jdbc.model.*;
+import org.apache.ignite.internal.processors.cache.*;
+import org.apache.ignite.internal.util.typedef.*;
+import org.apache.ignite.internal.util.typedef.internal.*;
+import org.apache.ignite.lang.*;
+import org.apache.ignite.testframework.*;
+import org.apache.ignite.testframework.junits.cache.*;
+import org.h2.jdbcx.*;
+import org.springframework.beans.*;
+import org.springframework.beans.factory.xml.*;
+import org.springframework.context.support.*;
+import org.springframework.core.io.*;
+
+import javax.cache.integration.*;
+import java.net.*;
+import java.sql.*;
+import java.util.*;
+import java.util.concurrent.*;
+
+/**
+ * Class for {@code PojoCacheStore} tests.
+ */
+public class CacheJdbcPojoStoreCompatibilityTest extends GridAbstractCacheStoreSelfTest<CacheJdbcPojoStore<Object,
Object>> {
+    /** DB connection URL. */
+    private static final String DFLT_CONN_URL = "jdbc:h2:mem:autoCacheStore;DB_CLOSE_DELAY=-1";
+
+    /** Default config with mapping. */
+    private static final String DFLT_MAPPING_CONFIG = "modules/core/src/test/config/store/jdbc/ignite-type-metadata.xml";
+
+    /** Organization count. */
+    protected static final int ORGANIZATION_CNT = 1000;
+
+    /** Person count. */
+    protected static final int PERSON_CNT = 100000;
+
+    /**
+     * @throws Exception If failed.
+     */
+    public CacheJdbcPojoStoreCompatibilityTest() throws Exception {
+        // No-op.
+    }
+
+    /** {@inheritDoc} */
+    @Override protected CacheJdbcPojoStore<Object, Object> store() {
+        CacheJdbcPojoStore<Object, Object> store = new CacheJdbcPojoStore<>();
+
+//        PGPoolingDataSource ds = new PGPoolingDataSource();
+//        ds.setUser("postgres");
+//        ds.setPassword("postgres");
+//        ds.setServerName("ip");
+//        ds.setDatabaseName("postgres");
+//        store.setDataSource(ds);
+
+//        MysqlDataSource ds = new MysqlDataSource();
+//        ds.setURL("jdbc:mysql://ip:port/dbname");
+//        ds.setUser("mysql");
+//        ds.setPassword("mysql");
+
+        store.setDataSource(JdbcConnectionPool.create(DFLT_CONN_URL, "sa", ""));
+
+        URL cfgUrl;
+
+        try {
+            cfgUrl = new URL(DFLT_MAPPING_CONFIG);
+        }
+        catch (MalformedURLException ignore) {
+            cfgUrl = U.resolveIgniteUrl(DFLT_MAPPING_CONFIG);
+        }
+
+        if (cfgUrl == null)
+            throw new IgniteException("Failed to resolve metadata path: " + DFLT_MAPPING_CONFIG);
+
+        try {
+            GenericApplicationContext springCtx = new GenericApplicationContext();
+
+            new XmlBeanDefinitionReader(springCtx).loadBeanDefinitions(new UrlResource(cfgUrl));
+
+            springCtx.refresh();
+
+            Collection<CacheJdbcPojoStoreType> typeMeta = springCtx.getBeansOfType(CacheJdbcPojoStoreType.class).values();
+
+            Map<Integer, Map<Object, CacheJdbcPojoStore.EntryMapping>> cacheMappings
= new HashMap<>();
+// TODO-1753 fix tests
+//            JdbcDialect dialect = store.resolveDialect();
+//
+//            GridTestUtils.setFieldValue(store, CacheJdbcPojoStore.class, "dialect", dialect);
+//
+//            Map<Object, CacheJdbcPojoStore.EntryMapping> entryMappings = U.newHashMap(typeMeta.size());
+//
+//            for (CacheJdbcPojoStoreType type : typeMeta)
+//                entryMappings.put(store.keyTypeId(type.getKeyType()),
+//                    new CacheJdbcPojoStore.EntryMapping(null, dialect, type));
+//
+//            store.prepareBuilders(null, typeMeta);
+//
+//            cacheMappings.put(null, entryMappings);
+//
+//            GridTestUtils.setFieldValue(store, CacheJdbcPojoStore.class, "cacheMappings",
cacheMappings);
+        }
+        catch (BeansException e) {
+            if (X.hasCause(e, ClassNotFoundException.class))
+                throw new IgniteException("Failed to instantiate Spring XML application context
" +
+                    "(make sure all classes used in Spring configuration are present at CLASSPATH)
" +
+                    "[springUrl=" + cfgUrl + ']', e);
+            else
+                throw new IgniteException("Failed to instantiate Spring XML application context
[springUrl=" +
+                    cfgUrl + ", err=" + e.getMessage() + ']', e);
+        }
+
+        return store;
+    }
+
+    /**
+     * @param store Store.
+     * @throws Exception If failed.
+     */
+    @Override protected void inject(CacheJdbcPojoStore<Object, Object> store) throws
Exception {
+        getTestResources().inject(store);
+
+        GridTestUtils.setFieldValue(store, CacheJdbcPojoStore.class, "ses", ses);
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void beforeTest() throws Exception {
+        Connection conn = store.openConnection(false);
+
+        Statement stmt = conn.createStatement();
+
+        try {
+            stmt.executeUpdate("delete from String_Entries");
+        }
+        catch (SQLException ignore) {
+            // No-op.
+        }
+
+        try {
+            stmt.executeUpdate("delete from UUID_Entries");
+        }
+        catch (SQLException ignore) {
+            // No-op.
+        }
+
+        try {
+            stmt.executeUpdate("delete from Organization");
+        }
+        catch (SQLException ignore) {
+            // No-op.
+        }
+
+        try {
+            stmt.executeUpdate("delete from Person");
+        }
+        catch (SQLException ignore) {
+            // No-op.
+        }
+
+        try {
+            stmt.executeUpdate("delete from Timestamp_Entries");
+        }
+        catch (SQLException ignore) {
+            // No-op.
+        }
+
+        stmt.executeUpdate("CREATE TABLE IF NOT EXISTS " +
+            "String_Entries (key varchar(100) not null, val varchar(100), PRIMARY KEY(key))");
+
+        stmt.executeUpdate("CREATE TABLE IF NOT EXISTS " +
+            "UUID_Entries (key binary(16) not null, val binary(16), PRIMARY KEY(key))");
+
+        stmt.executeUpdate("CREATE TABLE IF NOT EXISTS " +
+            "Timestamp_Entries (key timestamp not null, val integer, PRIMARY KEY(key))");
+
+        stmt.executeUpdate("CREATE TABLE IF NOT EXISTS " +
+            "Organization (id integer not null, name varchar(50), city varchar(50), PRIMARY
KEY(id))");
+
+        stmt.executeUpdate("CREATE TABLE IF NOT EXISTS " +
+            "Person (id integer not null, org_id integer, name varchar(50), PRIMARY KEY(id))");
+
+        stmt.executeUpdate("CREATE TABLE IF NOT EXISTS " +
+            "Person_Complex (id integer not null, org_id integer not null, city_id integer
not null, " +
+            "name varchar(50), salary integer, PRIMARY KEY(id))");
+
+        conn.commit();
+
+        U.closeQuiet(stmt);
+
+        U.closeQuiet(conn);
+
+        super.beforeTest();
+    }
+
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testLoadCache() throws Exception {
+        Connection conn = store.openConnection(false);
+
+        PreparedStatement orgStmt = conn.prepareStatement("INSERT INTO Organization(id, name,
city) VALUES (?, ?, ?)");
+
+        for (int i = 0; i < ORGANIZATION_CNT; i++) {
+            orgStmt.setInt(1, i);
+            orgStmt.setString(2, "name" + i);
+            orgStmt.setString(3, "city" + i % 10);
+
+            orgStmt.addBatch();
+        }
+
+        orgStmt.executeBatch();
+
+        U.closeQuiet(orgStmt);
+
+        conn.commit();
+
+        PreparedStatement prnStmt = conn.prepareStatement("INSERT INTO Person(id, org_id,
name) VALUES (?, ?, ?)");
+
+        for (int i = 0; i < PERSON_CNT; i++) {
+            prnStmt.setInt(1, i);
+            prnStmt.setInt(2, i % 100);
+            prnStmt.setString(3, "name" + i);
+
+            prnStmt.addBatch();
+        }
+
+        prnStmt.executeBatch();
+
+        conn.commit();
+
+        U.closeQuiet(prnStmt);
+
+        PreparedStatement prnComplexStmt = conn.prepareStatement("INSERT INTO Person_Complex(id,
org_id, city_id, name, salary) VALUES (?, ?, ?, ?, ?)");
+
+        for (int i = 0; i < PERSON_CNT; i++) {
+            prnComplexStmt.setInt(1, i);
+            prnComplexStmt.setInt(2, i % 500);
+            prnComplexStmt.setInt(3, i % 100);
+            prnComplexStmt.setString(4, "name" + i);
+
+            if (i > 0)
+                prnComplexStmt.setInt(5, 1000 + i * 500);
+            else // Add person with null salary
+                prnComplexStmt.setNull(5, java.sql.Types.INTEGER);
+
+            prnComplexStmt.addBatch();
+        }
+
+        prnComplexStmt.executeBatch();
+
+        U.closeQuiet(prnComplexStmt);
+
+        conn.commit();
+
+        U.closeQuiet(conn);
+
+        final Collection<OrganizationKey> orgKeys = new ConcurrentLinkedQueue<>();
+        final Collection<PersonKey> prnKeys = new ConcurrentLinkedQueue<>();
+        final Collection<PersonComplexKey> prnComplexKeys = new ConcurrentLinkedQueue<>();
+
+        IgniteBiInClosure<Object, Object> c = new CI2<Object, Object>() {
+            @Override public void apply(Object k, Object v) {
+                if (k instanceof OrganizationKey && v instanceof Organization)
+                    orgKeys.add((OrganizationKey)k);
+                else if (k instanceof PersonKey && v instanceof Person)
+                    prnKeys.add((PersonKey)k);
+                else if (k instanceof PersonComplexKey && v instanceof Person) {
+                    PersonComplexKey key = (PersonComplexKey)k;
+
+                    Person val = (Person)v;
+
+                    assert key.getId() == val.getId();
+                    assert key.getOrgId() == val.getOrgId();
+                    assertEquals("name"  + key.getId(), val.getName());
+
+                    prnComplexKeys.add((PersonComplexKey)k);
+                }
+            }
+        };
+
+        store.loadCache(c);
+
+        assertEquals(ORGANIZATION_CNT, orgKeys.size());
+        assertEquals(PERSON_CNT, prnKeys.size());
+        assertEquals(PERSON_CNT, prnComplexKeys.size());
+
+        Collection<OrganizationKey> tmpOrgKeys = new ArrayList<>(orgKeys);
+        Collection<PersonKey> tmpPrnKeys = new ArrayList<>(prnKeys);
+        Collection<PersonComplexKey> tmpPrnComplexKeys = new ArrayList<>(prnComplexKeys);
+
+        orgKeys.clear();
+        prnKeys.clear();
+        prnComplexKeys.clear();
+
+        store.loadCache(c, OrganizationKey.class.getName(), "SELECT name, city, id FROM ORGANIZATION",
+            PersonKey.class.getName(), "SELECT org_id, id, name FROM Person WHERE id <
1000");
+
+        assertEquals(ORGANIZATION_CNT, orgKeys.size());
+        assertEquals(1000, prnKeys.size());
+        assertEquals(0, prnComplexKeys.size());
+
+        store.deleteAll(tmpOrgKeys);
+        store.deleteAll(tmpPrnKeys);
+        store.deleteAll(tmpPrnComplexKeys);
+
+        orgKeys.clear();
+        prnKeys.clear();
+        prnComplexKeys.clear();
+
+        store.loadCache(c);
+
+        assertTrue(orgKeys.isEmpty());
+        assertTrue(prnKeys.isEmpty());
+        assertTrue(prnComplexKeys.isEmpty());
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testWriteRetry() throws Exception {
+        // Special dialect that will skip updates, to test write retry.
+        BasicJdbcDialect dialect = new BasicJdbcDialect() {
+            /** {@inheritDoc} */
+            @Override public String updateQuery(String tblName, Collection<String>
keyCols, Iterable<String> valCols) {
+                return super.updateQuery(tblName, keyCols, valCols) + " AND 1 = 0";
+            }
+        };
+
+        store.setDialect(dialect);
+
+        Map<String, Map<Object, CacheJdbcPojoStore.EntryMapping>> cacheMappings
=
+            GridTestUtils.getFieldValue(store, CacheJdbcPojoStore.class, "cacheMappings");
+
+        CacheJdbcPojoStore.EntryMapping em = cacheMappings.get(null).get(OrganizationKey.class);
+
+        CacheJdbcPojoStoreType typeMeta = GridTestUtils.getFieldValue(em, CacheJdbcPojoStore.EntryMapping.class,
"typeMeta");
+
+        cacheMappings.get(null).put(OrganizationKey.class,
+            new CacheJdbcPojoStore.EntryMapping(null, dialect, typeMeta));
+
+        Connection conn = store.openConnection(false);
+
+        PreparedStatement orgStmt = conn.prepareStatement("INSERT INTO Organization(id, name,
city) VALUES (?, ?, ?)");
+
+        orgStmt.setInt(1, 1);
+        orgStmt.setString(2, "name" + 1);
+        orgStmt.setString(3, "city" + 1);
+
+        orgStmt.executeUpdate();
+
+        U.closeQuiet(orgStmt);
+
+        conn.commit();
+
+        OrganizationKey k1 = new OrganizationKey(1);
+        Organization v1 = new Organization(1, "Name1", "City1");
+
+        ses.newSession(null);
+
+        try {
+            store.write(new CacheEntryImpl<>(k1, v1));
+        }
+        catch (CacheWriterException e) {
+            if (!e.getMessage().startsWith("Failed insert entry in database, violate a unique
index or primary key") ||
+                e.getSuppressed().length != 2)
+                throw e;
+        }
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testTimestamp() throws Exception {
+        Timestamp k = new Timestamp(System.currentTimeMillis());
+
+        ses.newSession(null);
+
+        Integer v = 5;
+
+        store.write(new CacheEntryImpl<>(k, v));
+
+        assertEquals(v, store.load(k));
+
+        store.delete(k);
+
+        assertNull(store.load(k));
+    }
+}

http://git-wip-us.apache.org/repos/asf/ignite/blob/8e26c84d/modules/core/src/test/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStoreTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStoreTest.java
b/modules/core/src/test/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStoreTest.java
index 746059d..af6f541 100644
--- a/modules/core/src/test/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStoreTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStoreTest.java
@@ -17,22 +17,21 @@
 
 package org.apache.ignite.cache.store.jdbc;
 
-import java.net.MalformedURLException;
-import java.net.URL;
 import java.sql.Connection;
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
 import java.sql.Statement;
 import java.sql.Timestamp;
+import java.sql.Types;
 import java.util.ArrayList;
 import java.util.Collection;
-import java.util.HashMap;
 import java.util.Map;
+import java.util.UUID;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import javax.cache.integration.CacheWriterException;
-import org.apache.ignite.IgniteException;
+
 import org.apache.ignite.cache.store.jdbc.dialect.BasicJdbcDialect;
-import org.apache.ignite.cache.store.jdbc.dialect.JdbcDialect;
+import org.apache.ignite.cache.store.jdbc.dialect.H2Dialect;
 import org.apache.ignite.cache.store.jdbc.model.Organization;
 import org.apache.ignite.cache.store.jdbc.model.OrganizationKey;
 import org.apache.ignite.cache.store.jdbc.model.Person;
@@ -40,17 +39,11 @@ import org.apache.ignite.cache.store.jdbc.model.PersonComplexKey;
 import org.apache.ignite.cache.store.jdbc.model.PersonKey;
 import org.apache.ignite.internal.processors.cache.CacheEntryImpl;
 import org.apache.ignite.internal.util.typedef.CI2;
-import org.apache.ignite.internal.util.typedef.F;
-import org.apache.ignite.internal.util.typedef.X;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.lang.IgniteBiInClosure;
 import org.apache.ignite.testframework.GridTestUtils;
 import org.apache.ignite.testframework.junits.cache.GridAbstractCacheStoreSelfTest;
 import org.h2.jdbcx.JdbcConnectionPool;
-import org.springframework.beans.BeansException;
-import org.springframework.beans.factory.xml.XmlBeanDefinitionReader;
-import org.springframework.context.support.GenericApplicationContext;
-import org.springframework.core.io.UrlResource;
 
 /**
  * Class for {@code PojoCacheStore} tests.
@@ -77,7 +70,93 @@ public class CacheJdbcPojoStoreTest extends GridAbstractCacheStoreSelfTest<Cache
 
     /** {@inheritDoc} */
     @Override protected CacheJdbcPojoStore<Object, Object> store() {
-        CacheJdbcPojoStore<Object, Object> store = new CacheJdbcPojoStore<>();
+        CacheJdbcPojoStoreFactory<Object, Object> storeFactory = new CacheJdbcPojoStoreFactory<>();
+
+        CacheJdbcPojoStoreConfiguration storeCfg = new CacheJdbcPojoStoreConfiguration();
+        storeCfg.setDialect(new H2Dialect());
+
+        CacheJdbcPojoStoreType[] storeTypes = new CacheJdbcPojoStoreType[6];
+
+        storeTypes[0] = new CacheJdbcPojoStoreType();
+        storeTypes[0].setDatabaseSchema("PUBLIC");
+        storeTypes[0].setDatabaseTable("ORGANIZATION");
+        storeTypes[0].setKeyType("org.apache.ignite.cache.store.jdbc.model.OrganizationKey");
+        storeTypes[0].setKeyFields(new CacheJdbcPojoStoreTypeField[1]);
+        storeTypes[0].getKeyFields()[0] = new CacheJdbcPojoStoreTypeField(Types.INTEGER,
"ID", Integer.class, "id");
+
+        storeTypes[0].setValueType("org.apache.ignite.cache.store.jdbc.model.Organization");
+        storeTypes[0].setValueFields(new CacheJdbcPojoStoreTypeField[3]);
+        storeTypes[0].getValueFields()[0] = new CacheJdbcPojoStoreTypeField(Types.INTEGER,
"ID", Integer.class, "id");
+        storeTypes[0].getValueFields()[1] = new CacheJdbcPojoStoreTypeField(Types.VARCHAR,
"NAME", String.class, "name");
+        storeTypes[0].getValueFields()[2] = new CacheJdbcPojoStoreTypeField(Types.VARCHAR,
"CITY", String.class, "city");
+
+        storeTypes[1] = new CacheJdbcPojoStoreType();
+        storeTypes[1].setDatabaseSchema("PUBLIC");
+        storeTypes[1].setDatabaseTable("PERSON");
+        storeTypes[1].setKeyType("org.apache.ignite.cache.store.jdbc.model.PersonKey");
+        storeTypes[1].setKeyFields(new CacheJdbcPojoStoreTypeField[1]);
+        storeTypes[1].getKeyFields()[0] = new CacheJdbcPojoStoreTypeField(Types.INTEGER,
"ID", Integer.class, "id");
+
+        storeTypes[1].setValueType("org.apache.ignite.cache.store.jdbc.model.Person");
+        storeTypes[1].setValueFields(new CacheJdbcPojoStoreTypeField[3]);
+        storeTypes[1].getValueFields()[0] = new CacheJdbcPojoStoreTypeField(Types.INTEGER,
"ID", Integer.class, "id");
+        storeTypes[1].getValueFields()[1] = new CacheJdbcPojoStoreTypeField(Types.INTEGER,
"ORG_ID", Integer.class, "orgId");
+        storeTypes[1].getValueFields()[2] = new CacheJdbcPojoStoreTypeField(Types.VARCHAR,
"NAME", String.class, "name");
+
+        storeTypes[2] = new CacheJdbcPojoStoreType();
+        storeTypes[2].setDatabaseSchema("PUBLIC");
+        storeTypes[2].setDatabaseTable("PERSON_COMPLEX");
+        storeTypes[2].setKeyType("org.apache.ignite.cache.store.jdbc.model.PersonComplexKey");
+        storeTypes[2].setKeyFields(new CacheJdbcPojoStoreTypeField[3]);
+        storeTypes[2].getKeyFields()[0] = new CacheJdbcPojoStoreTypeField(Types.INTEGER,
"ID", int.class, "id");
+        storeTypes[2].getKeyFields()[1] = new CacheJdbcPojoStoreTypeField(Types.INTEGER,
"ORG_ID", int.class, "orgId");
+        storeTypes[2].getKeyFields()[2] = new CacheJdbcPojoStoreTypeField(Types.INTEGER,
"CITY_ID", int.class, "cityId");
+
+        storeTypes[2].setValueType("org.apache.ignite.cache.store.jdbc.model.Person");
+        storeTypes[2].setValueFields(new CacheJdbcPojoStoreTypeField[4]);
+        storeTypes[2].getValueFields()[0] = new CacheJdbcPojoStoreTypeField(Types.INTEGER,
"ID", Integer.class, "id");
+        storeTypes[2].getValueFields()[1] = new CacheJdbcPojoStoreTypeField(Types.INTEGER,
"ORG_ID", Integer.class, "orgId");
+        storeTypes[2].getValueFields()[2] = new CacheJdbcPojoStoreTypeField(Types.VARCHAR,
"NAME", String.class, "name");
+        storeTypes[2].getValueFields()[3] = new CacheJdbcPojoStoreTypeField(Types.VARCHAR,
"SALARY", Integer.class, "salary");
+
+        storeTypes[3] = new CacheJdbcPojoStoreType();
+        storeTypes[3].setDatabaseSchema("PUBLIC");
+        storeTypes[3].setDatabaseTable("TIMESTAMP_ENTRIES");
+        storeTypes[3].setKeyType("java.sql.Timestamp");
+        storeTypes[3].setKeyFields(new CacheJdbcPojoStoreTypeField[1]);
+        storeTypes[3].getKeyFields()[0] = new CacheJdbcPojoStoreTypeField(Types.TIMESTAMP,
"KEY", Timestamp.class, null);
+
+        storeTypes[3].setValueType("java.lang.Integer");
+        storeTypes[3].setValueFields(new CacheJdbcPojoStoreTypeField[1]);
+        storeTypes[3].getValueFields()[0] = new CacheJdbcPojoStoreTypeField(Types.INTEGER,
"VAL", Integer.class, null);
+
+        storeTypes[4] = new CacheJdbcPojoStoreType();
+        storeTypes[4].setDatabaseSchema("PUBLIC");
+        storeTypes[4].setDatabaseTable("STRING_ENTRIES");
+        storeTypes[4].setKeyType("java.lang.String");
+        storeTypes[4].setKeyFields(new CacheJdbcPojoStoreTypeField[1]);
+        storeTypes[4].getKeyFields()[0] = new CacheJdbcPojoStoreTypeField(Types.VARCHAR,
"KEY", String.class, null);
+
+        storeTypes[4].setValueType("java.lang.String");
+        storeTypes[4].setValueFields(new CacheJdbcPojoStoreTypeField[1]);
+        storeTypes[4].getValueFields()[0] = new CacheJdbcPojoStoreTypeField(Types.VARCHAR,
"VAL", Integer.class, null);
+
+        storeTypes[5] = new CacheJdbcPojoStoreType();
+        storeTypes[5].setDatabaseSchema("PUBLIC");
+        storeTypes[5].setDatabaseTable("UUID_ENTRIES");
+        storeTypes[5].setKeyType("java.util.UUID");
+        storeTypes[5].setKeyFields(new CacheJdbcPojoStoreTypeField[1]);
+        storeTypes[5].getKeyFields()[0] = new CacheJdbcPojoStoreTypeField(Types.BINARY, "KEY",
UUID.class, null);
+
+        storeTypes[5].setValueType("java.util.UUID");
+        storeTypes[5].setValueFields(new CacheJdbcPojoStoreTypeField[1]);
+        storeTypes[5].getValueFields()[0] = new CacheJdbcPojoStoreTypeField(Types.BINARY,
"VAL", UUID.class, null);
+
+        storeCfg.setTypes(storeTypes);
+
+        storeFactory.setConfiguration(storeCfg);
+
+        CacheJdbcPojoStore<Object, Object> store = storeFactory.create();
 
 //        PGPoolingDataSource ds = new PGPoolingDataSource();
 //        ds.setUser("postgres");
@@ -91,57 +170,9 @@ public class CacheJdbcPojoStoreTest extends GridAbstractCacheStoreSelfTest<Cache
 //        ds.setUser("mysql");
 //        ds.setPassword("mysql");
 
+        // H2 DataSource
         store.setDataSource(JdbcConnectionPool.create(DFLT_CONN_URL, "sa", ""));
 
-        URL cfgUrl;
-
-        try {
-            cfgUrl = new URL(DFLT_MAPPING_CONFIG);
-        }
-        catch (MalformedURLException ignore) {
-            cfgUrl = U.resolveIgniteUrl(DFLT_MAPPING_CONFIG);
-        }
-
-        if (cfgUrl == null)
-            throw new IgniteException("Failed to resolve metadata path: " + DFLT_MAPPING_CONFIG);
-
-        try {
-            GenericApplicationContext springCtx = new GenericApplicationContext();
-
-            new XmlBeanDefinitionReader(springCtx).loadBeanDefinitions(new UrlResource(cfgUrl));
-
-            springCtx.refresh();
-
-            Collection<CacheJdbcPojoStoreType> typeMeta = springCtx.getBeansOfType(CacheJdbcPojoStoreType.class).values();
-
-            Map<Integer, Map<Object, CacheJdbcPojoStore.EntryMapping>> cacheMappings
= new HashMap<>();
-
-            JdbcDialect dialect = store.resolveDialect();
-
-            GridTestUtils.setFieldValue(store, CacheJdbcPojoStore.class, "dialect", dialect);
-
-            Map<Object, CacheJdbcPojoStore.EntryMapping> entryMappings = U.newHashMap(typeMeta.size());
-
-            for (CacheJdbcPojoStoreType type : typeMeta)
-                entryMappings.put(store.keyTypeId(type.getKeyType()),
-                    new CacheJdbcPojoStore.EntryMapping(null, dialect, type));
-
-            store.prepareBuilders(null, typeMeta);
-
-            cacheMappings.put(null, entryMappings);
-
-            GridTestUtils.setFieldValue(store, CacheJdbcPojoStore.class, "cacheMappings",
cacheMappings);
-        }
-        catch (BeansException e) {
-            if (X.hasCause(e, ClassNotFoundException.class))
-                throw new IgniteException("Failed to instantiate Spring XML application context
" +
-                    "(make sure all classes used in Spring configuration are present at CLASSPATH)
" +
-                    "[springUrl=" + cfgUrl + ']', e);
-            else
-                throw new IgniteException("Failed to instantiate Spring XML application context
[springUrl=" +
-                    cfgUrl + ", err=" + e.getMessage() + ']', e);
-        }
-
         return store;
     }
 
@@ -224,7 +255,6 @@ public class CacheJdbcPojoStoreTest extends GridAbstractCacheStoreSelfTest<Cache
         super.beforeTest();
     }
 
-
     /**
      * @throws Exception If failed.
      */
@@ -274,7 +304,7 @@ public class CacheJdbcPojoStoreTest extends GridAbstractCacheStoreSelfTest<Cache
             if (i > 0)
                 prnComplexStmt.setInt(5, 1000 + i * 500);
             else // Add person with null salary
-                prnComplexStmt.setNull(5, java.sql.Types.INTEGER);
+                prnComplexStmt.setNull(5, Types.INTEGER);
 
             prnComplexStmt.addBatch();
         }
@@ -302,9 +332,9 @@ public class CacheJdbcPojoStoreTest extends GridAbstractCacheStoreSelfTest<Cache
 
                     Person val = (Person)v;
 
-                    assert key.getId() == val.getId();
-                    assert key.getOrgId() == val.getOrgId();
-                    assertEquals("name"  + key.getId(), val.getName());
+                    assertTrue("Key ID should be the same as value ID", key.getId() == val.getId());
+                    assertTrue("Key orgID should be the same as value orgID", key.getOrgId()
== val.getOrgId());
+                    assertEquals("name" + key.getId(), val.getName());
 
                     prnComplexKeys.add((PersonComplexKey)k);
                 }


Mime
View raw message