You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ak...@apache.org on 2015/10/23 18:24:41 UTC
ignite git commit: IGNITE-1753 WIP on POJO store refactoring + fixed
first POJO store test.
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 <ak...@gridgain.com>
Authored: Fri Oct 23 23:24:32 2015 +0700
Committer: AKuznetsov <ak...@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);
}