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/28 09:19:09 UTC
[1/4] ignite git commit: IGNITE-1753 Wip reworking code.
Repository: ignite
Updated Branches:
refs/heads/ignite-1753-1282 8d325a4c4 -> 852e54c06
http://git-wip-us.apache.org/repos/asf/ignite/blob/2961aaa7/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStoreTypeField.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStoreTypeField.java b/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStoreTypeField.java
deleted file mode 100644
index ccd26b3..0000000
--- a/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStoreTypeField.java
+++ /dev/null
@@ -1,172 +0,0 @@
-/*
- * 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 java.io.Serializable;
-import org.apache.ignite.internal.util.typedef.internal.S;
-
-/**
- * Description of how field declared in database and in cache.
- */
-public class CacheJdbcPojoStoreTypeField implements Serializable {
- /** */
- private static final long serialVersionUID = 0L;
-
- /** Field JDBC type in database. */
- private int dbFieldType;
-
- /** Field name in database. */
- private String dbFieldName;
-
- /** Field java type. */
- private Class<?> javaFieldType;
-
- /** Field name in java object. */
- private String javaFieldName;
-
- /**
- * Default constructor.
- */
- public CacheJdbcPojoStoreTypeField() {
- // No-op.
- }
-
- /**
- * Full constructor.
- *
- * @param dbFieldType Field JDBC type in database.
- * @param dbFieldName Field name in database.
- * @param javaFieldType Field java type.
- * @param javaFieldName Field name in java object.
- */
- public CacheJdbcPojoStoreTypeField(int dbFieldType, String dbFieldName, Class<?> javaFieldType, String javaFieldName) {
- this.dbFieldType = dbFieldType;
- this.dbFieldName = dbFieldName;
- this.javaFieldType = javaFieldType;
- this.javaFieldName = javaFieldName;
- }
-
- /**
- * Copy constructor.
- *
- * @param field Field to copy.
- */
- public CacheJdbcPojoStoreTypeField(CacheJdbcPojoStoreTypeField field) {
- this(field.getDatabaseFieldType(), field.getDatabaseFieldName(),
- field.getJavaFieldType(), field.getJavaFieldName());
- }
-
- /**
- * @return Column JDBC type in database.
- */
- public int getDatabaseFieldType() {
- return dbFieldType;
- }
-
- /**
- * @param dbType Column JDBC type in database.
- * @return {@code this} for chaining.
- */
- public CacheJdbcPojoStoreTypeField setDatabaseFieldType(int dbType) {
- this.dbFieldType = dbType;
-
- return this;
- }
-
-
- /**
- * @return Column name in database.
- */
- public String getDatabaseFieldName() {
- return dbFieldName;
- }
-
- /**
- * @param dbName Column name in database.
- * @return {@code this} for chaining.
- */
- public CacheJdbcPojoStoreTypeField setDatabaseFieldName(String dbName) {
- this.dbFieldName = dbName;
-
- return this;
- }
-
- /**
- * @return Field java type.
- */
- public Class<?> getJavaFieldType() {
- return javaFieldType;
- }
-
- /**
- * @param javaType Corresponding java type.
- * @return {@code this} for chaining.
- */
- public CacheJdbcPojoStoreTypeField setJavaFieldType(Class<?> javaType) {
- this.javaFieldType = javaType;
-
- return this;
- }
-
- /**
- * @return Field name in java object.
- */
- public String getJavaFieldName() {
- return javaFieldName;
- }
-
- /**
- * @param javaName Field name in java object.
- * @return {@code this} for chaining.
- */
- public CacheJdbcPojoStoreTypeField setJavaFieldName(String javaName) {
- this.javaFieldName = javaName;
-
- return this;
- }
-
- /** {@inheritDoc} */
- @Override public boolean equals(Object o) {
- if (this == o)
- return true;
-
- if (!(o instanceof CacheJdbcPojoStoreTypeField))
- return false;
-
- CacheJdbcPojoStoreTypeField that = (CacheJdbcPojoStoreTypeField)o;
-
- return dbFieldType == that.dbFieldType && dbFieldName.equals(that.dbFieldName) &&
- javaFieldType == that.javaFieldType && javaFieldName.equals(that.javaFieldName);
- }
-
- /** {@inheritDoc} */
- @Override public int hashCode() {
- int res = dbFieldType;
- res = 31 * res + dbFieldName.hashCode();
-
- res = 31 * res + javaFieldType.hashCode();
- res = 31 * res + javaFieldName.hashCode();
-
- return res;
- }
-
- /** {@inheritDoc} */
- @Override public String toString() {
- return S.toString(CacheJdbcPojoStoreTypeField.class, this);
- }
-}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2961aaa7/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/JdbcType.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/JdbcType.java b/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/JdbcType.java
new file mode 100644
index 0000000..cb08fc1
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/JdbcType.java
@@ -0,0 +1,272 @@
+/*
+ * 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 java.io.Serializable;
+import org.apache.ignite.internal.util.tostring.GridToStringInclude;
+
+/**
+ * Description for type that could be stored into database by store.
+ */
+public class JdbcType implements Serializable {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** Cache name. */
+ private String cacheName;
+
+ /** Schema name in database. */
+ private String dbSchema;
+
+ /** Table name in database. */
+ private String dbTbl;
+
+ /** Key class used to store key in cache. */
+ private String keyType;
+
+ /** List of fields descriptors for key object. */
+ @GridToStringInclude
+ private JdbcTypeField[] keyFields;
+
+ /** Value class used to store value in cache. */
+ private String valType;
+
+ /** List of fields descriptors for value object. */
+ @GridToStringInclude
+ private JdbcTypeField[] valFields;
+
+ /** If {@code true} object is stored as IgniteObject. */
+ private boolean keepSerialized;
+
+ /**
+ * Empty constructor (all values are initialized to their defaults).
+ */
+ public JdbcType() {
+ /* No-op. */
+ }
+
+ /**
+ * Copy constructor.
+ *
+ * @param type Type to copy.
+ */
+ public JdbcType(JdbcType type) {
+ cacheName = type.getCacheName();
+
+ dbSchema = type.getDatabaseSchema();
+ dbTbl = type.getDatabaseTable();
+
+ keyType = type.getKeyType();
+ keyFields = type.getKeyFields();
+
+ valType = type.getValueType();
+ valFields = type.getValueFields();
+
+ keepSerialized = type.isKeepSerialized();
+ }
+
+ /**
+ * Gets associated cache name.
+ *
+ * @return Сache name.
+ */
+ public String getCacheName() {
+ return cacheName;
+ }
+
+ /**
+ * Sets associated cache name.
+ *
+ * @param cacheName Cache name.
+ */
+ public JdbcType setCacheName(String cacheName) {
+ this.cacheName = cacheName;
+
+ return this;
+ }
+
+ /**
+ * Gets database schema name.
+ *
+ * @return Schema name.
+ */
+ public String getDatabaseSchema() {
+ return dbSchema;
+ }
+
+ /**
+ * Sets database schema name.
+ *
+ * @param dbSchema Schema name.
+ */
+ public JdbcType setDatabaseSchema(String dbSchema) {
+ this.dbSchema = dbSchema;
+
+ return this;
+ }
+
+ /**
+ * Gets table name in database.
+ *
+ * @return Table name in database.
+ */
+ public String getDatabaseTable() {
+ return dbTbl;
+ }
+
+ /**
+ * Table name in database.
+ *
+ * @param dbTbl Table name in database.
+ * @return {@code this} for chaining.
+ */
+ public JdbcType setDatabaseTable(String dbTbl) {
+ this.dbTbl = dbTbl;
+
+ return this;
+ }
+
+ /**
+ * Gets key type.
+ *
+ * @return Key type.
+ */
+ public String getKeyType() {
+ return keyType;
+ }
+
+ /**
+ * Sets key type.
+ *
+ * @param keyType Key type.
+ * @return {@code this} for chaining.
+ */
+ public JdbcType setKeyType(String keyType) {
+ this.keyType = keyType;
+
+ return this;
+ }
+
+ /**
+ * Sets key type.
+ *
+ * @param cls Key type class.
+ * @return {@code this} for chaining.
+ */
+ public JdbcType setKeyType(Class<?> cls) {
+ setKeyType(cls.getName());
+
+ return this;
+ }
+
+ /**
+ * Gets value type.
+ *
+ * @return Key type.
+ */
+ public String getValueType() {
+ return valType;
+ }
+
+ /**
+ * Sets value type.
+ *
+ * @param valType Value type.
+ * @return {@code this} for chaining.
+ */
+ public JdbcType setValueType(String valType) {
+ this.valType = valType;
+
+ return this;
+ }
+
+ /**
+ * Sets value type.
+ *
+ * @param cls Value type class.
+ * @return {@code this} for chaining.
+ */
+ public JdbcType setValueType(Class<?> cls) {
+ setValueType(cls.getName());
+
+ return this;
+ }
+
+ /**
+ * Gets optional persistent key fields (needed only if {@link CacheJdbcPojoStore} is used).
+ *
+ * @return Persistent key fields.
+ */
+ public JdbcTypeField[] getKeyFields() {
+ return keyFields;
+ }
+
+ /**
+ * Sets optional persistent key fields (needed only if {@link CacheJdbcPojoStore} is used).
+ *
+ * @param keyFields Persistent key fields.
+ * @return {@code this} for chaining.
+ */
+ public JdbcType setKeyFields(JdbcTypeField... keyFields) {
+ this.keyFields = keyFields;
+
+ return this;
+ }
+
+ /**
+ * Gets optional persistent value fields (needed only if {@link CacheJdbcPojoStore} is used).
+ *
+ * @return Persistent value fields.
+ */
+ public JdbcTypeField[] getValueFields() {
+ return valFields;
+ }
+
+ /**
+ * Sets optional persistent value fields (needed only if {@link CacheJdbcPojoStore} is used).
+ *
+ * @param valFields Persistent value fields.
+ * @return {@code this} for chaining.
+ */
+ public JdbcType setValueFields(JdbcTypeField... valFields) {
+ this.valFields = valFields;
+
+ return this;
+ }
+
+ /**
+ * Gets how value stored in cache.
+ *
+ * @return {@code true} if object is stored as IgniteObject.
+ */
+ public boolean isKeepSerialized() {
+ return keepSerialized;
+ }
+
+ /**
+ * Sets how value stored in cache.
+ *
+ * @param keepSerialized {@code true} if object is stored as IgniteObject.
+ * @return {@code this} for chaining.
+ */
+ public JdbcType setKeepSerialized(boolean keepSerialized) {
+ this.keepSerialized = keepSerialized;
+
+ return this;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2961aaa7/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/JdbcTypeField.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/JdbcTypeField.java b/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/JdbcTypeField.java
new file mode 100644
index 0000000..ec8c18d
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/JdbcTypeField.java
@@ -0,0 +1,172 @@
+/*
+ * 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 java.io.Serializable;
+import org.apache.ignite.internal.util.typedef.internal.S;
+
+/**
+ * Description of how field declared in database and in cache.
+ */
+public class JdbcTypeField implements Serializable {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** Field JDBC type in database. */
+ private int dbFieldType;
+
+ /** Field name in database. */
+ private String dbFieldName;
+
+ /** Field java type. */
+ private Class<?> javaFieldType;
+
+ /** Field name in java object. */
+ private String javaFieldName;
+
+ /**
+ * Default constructor.
+ */
+ public JdbcTypeField() {
+ // No-op.
+ }
+
+ /**
+ * Full constructor.
+ *
+ * @param dbFieldType Field JDBC type in database.
+ * @param dbFieldName Field name in database.
+ * @param javaFieldType Field java type.
+ * @param javaFieldName Field name in java object.
+ */
+ public JdbcTypeField(int dbFieldType, String dbFieldName, Class<?> javaFieldType, String javaFieldName) {
+ this.dbFieldType = dbFieldType;
+ this.dbFieldName = dbFieldName;
+ this.javaFieldType = javaFieldType;
+ this.javaFieldName = javaFieldName;
+ }
+
+ /**
+ * Copy constructor.
+ *
+ * @param field Field to copy.
+ */
+ public JdbcTypeField(JdbcTypeField field) {
+ this(field.getDatabaseFieldType(), field.getDatabaseFieldName(),
+ field.getJavaFieldType(), field.getJavaFieldName());
+ }
+
+ /**
+ * @return Column JDBC type in database.
+ */
+ public int getDatabaseFieldType() {
+ return dbFieldType;
+ }
+
+ /**
+ * @param dbType Column JDBC type in database.
+ * @return {@code this} for chaining.
+ */
+ public JdbcTypeField setDatabaseFieldType(int dbType) {
+ this.dbFieldType = dbType;
+
+ return this;
+ }
+
+
+ /**
+ * @return Column name in database.
+ */
+ public String getDatabaseFieldName() {
+ return dbFieldName;
+ }
+
+ /**
+ * @param dbName Column name in database.
+ * @return {@code this} for chaining.
+ */
+ public JdbcTypeField setDatabaseFieldName(String dbName) {
+ this.dbFieldName = dbName;
+
+ return this;
+ }
+
+ /**
+ * @return Field java type.
+ */
+ public Class<?> getJavaFieldType() {
+ return javaFieldType;
+ }
+
+ /**
+ * @param javaType Corresponding java type.
+ * @return {@code this} for chaining.
+ */
+ public JdbcTypeField setJavaFieldType(Class<?> javaType) {
+ this.javaFieldType = javaType;
+
+ return this;
+ }
+
+ /**
+ * @return Field name in java object.
+ */
+ public String getJavaFieldName() {
+ return javaFieldName;
+ }
+
+ /**
+ * @param javaName Field name in java object.
+ * @return {@code this} for chaining.
+ */
+ public JdbcTypeField setJavaFieldName(String javaName) {
+ this.javaFieldName = javaName;
+
+ return this;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ if (this == o)
+ return true;
+
+ if (!(o instanceof JdbcTypeField))
+ return false;
+
+ JdbcTypeField that = (JdbcTypeField)o;
+
+ return dbFieldType == that.dbFieldType && dbFieldName.equals(that.dbFieldName) &&
+ javaFieldType == that.javaFieldType && javaFieldName.equals(that.javaFieldName);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ int res = dbFieldType;
+ res = 31 * res + dbFieldName.hashCode();
+
+ res = 31 * res + javaFieldType.hashCode();
+ res = 31 * res + javaFieldName.hashCode();
+
+ return res;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(JdbcTypeField.class, this);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2961aaa7/modules/core/src/test/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStoreSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStoreSelfTest.java b/modules/core/src/test/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStoreSelfTest.java
index 9f50507..7566572 100644
--- a/modules/core/src/test/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStoreSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStoreSelfTest.java
@@ -20,7 +20,6 @@ package org.apache.ignite.cache.store.jdbc;
import java.sql.Types;
import org.apache.ignite.marshaller.Marshaller;
import org.apache.ignite.marshaller.optimized.OptimizedMarshaller;
-import org.apache.ignite.marshaller.portable.PortableMarshaller;
/**
* Class for {@code PojoCacheStore} tests.
@@ -28,36 +27,34 @@ import org.apache.ignite.marshaller.portable.PortableMarshaller;
public class CacheJdbcPojoStoreSelfTest extends CacheJdbcStoreAbstractSelfTest {
/** {@inheritDoc} */
@Override protected Marshaller marshaller(){
- OptimizedMarshaller marsh = new OptimizedMarshaller();
-
- return marsh;
+ return new OptimizedMarshaller();
}
/** {@inheritDoc} */
- @Override protected CacheJdbcPojoStoreType[] storeTypes() {
- CacheJdbcPojoStoreType[] storeTypes = new CacheJdbcPojoStoreType[2];
+ @Override protected JdbcType[] storeTypes() {
+ JdbcType[] storeTypes = new JdbcType[2];
- storeTypes[0] = new CacheJdbcPojoStoreType();
+ storeTypes[0] = new JdbcType();
storeTypes[0].setDatabaseSchema("PUBLIC");
storeTypes[0].setDatabaseTable("ORGANIZATION");
storeTypes[0].setKeyType("org.apache.ignite.cache.store.jdbc.model.OrganizationKey");
- storeTypes[0].setKeyFields(new CacheJdbcPojoStoreTypeField(Types.INTEGER, "ID", Integer.class, "id"));
+ storeTypes[0].setKeyFields(new JdbcTypeField(Types.INTEGER, "ID", Integer.class, "id"));
storeTypes[0].setValueType("org.apache.ignite.cache.store.jdbc.model.Organization");
storeTypes[0].setValueFields(
- new CacheJdbcPojoStoreTypeField(Types.INTEGER, "ID", Integer.class, "id"),
- new CacheJdbcPojoStoreTypeField(Types.VARCHAR, "NAME", String.class, "name"),
- new CacheJdbcPojoStoreTypeField(Types.VARCHAR, "CITY", String.class, "city"));
+ new JdbcTypeField(Types.INTEGER, "ID", Integer.class, "id"),
+ new JdbcTypeField(Types.VARCHAR, "NAME", String.class, "name"),
+ new JdbcTypeField(Types.VARCHAR, "CITY", String.class, "city"));
- storeTypes[1] = new CacheJdbcPojoStoreType();
+ storeTypes[1] = new JdbcType();
storeTypes[1].setDatabaseSchema("PUBLIC");
storeTypes[1].setDatabaseTable("PERSON");
storeTypes[1].setKeyType("org.apache.ignite.cache.store.jdbc.model.PersonKey");
- storeTypes[1].setKeyFields(new CacheJdbcPojoStoreTypeField(Types.INTEGER, "ID", Integer.class, "id"));
+ storeTypes[1].setKeyFields(new JdbcTypeField(Types.INTEGER, "ID", Integer.class, "id"));
storeTypes[1].setValueType("org.apache.ignite.cache.store.jdbc.model.Person");
storeTypes[1].setValueFields(
- new CacheJdbcPojoStoreTypeField(Types.INTEGER, "ID", Integer.class, "id"),
- new CacheJdbcPojoStoreTypeField(Types.INTEGER, "ORG_ID", Integer.class, "orgId"),
- new CacheJdbcPojoStoreTypeField(Types.VARCHAR, "NAME", String.class, "name"));
+ new JdbcTypeField(Types.INTEGER, "ID", Integer.class, "id"),
+ new JdbcTypeField(Types.INTEGER, "ORG_ID", Integer.class, "orgId"),
+ new JdbcTypeField(Types.VARCHAR, "NAME", String.class, "name"));
return storeTypes;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2961aaa7/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 11c0040..8c4f7d4 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
@@ -70,74 +70,74 @@ public class CacheJdbcPojoStoreTest extends GridAbstractCacheStoreSelfTest<Cache
CacheJdbcPojoStoreConfiguration storeCfg = new CacheJdbcPojoStoreConfiguration();
storeCfg.setDialect(new H2Dialect());
- CacheJdbcPojoStoreType[] storeTypes = new CacheJdbcPojoStoreType[6];
+ JdbcType[] storeTypes = new JdbcType[6];
- storeTypes[0] = new CacheJdbcPojoStoreType();
+ storeTypes[0] = new JdbcType();
storeTypes[0].setDatabaseSchema("PUBLIC");
storeTypes[0].setDatabaseTable("ORGANIZATION");
storeTypes[0].setKeyType("org.apache.ignite.cache.store.jdbc.model.OrganizationKey");
- storeTypes[0].setKeyFields(new CacheJdbcPojoStoreTypeField(Types.INTEGER, "ID", Integer.class, "id"));
+ storeTypes[0].setKeyFields(new JdbcTypeField(Types.INTEGER, "ID", Integer.class, "id"));
storeTypes[0].setValueType("org.apache.ignite.cache.store.jdbc.model.Organization");
storeTypes[0].setValueFields(
- new CacheJdbcPojoStoreTypeField(Types.INTEGER, "ID", Integer.class, "id"),
- new CacheJdbcPojoStoreTypeField(Types.VARCHAR, "NAME", String.class, "name"),
- new CacheJdbcPojoStoreTypeField(Types.VARCHAR, "CITY", String.class, "city"));
+ new JdbcTypeField(Types.INTEGER, "ID", Integer.class, "id"),
+ new JdbcTypeField(Types.VARCHAR, "NAME", String.class, "name"),
+ new JdbcTypeField(Types.VARCHAR, "CITY", String.class, "city"));
- storeTypes[1] = new CacheJdbcPojoStoreType();
+ storeTypes[1] = new JdbcType();
storeTypes[1].setDatabaseSchema("PUBLIC");
storeTypes[1].setDatabaseTable("PERSON");
storeTypes[1].setKeyType("org.apache.ignite.cache.store.jdbc.model.PersonKey");
- storeTypes[1].setKeyFields(new CacheJdbcPojoStoreTypeField(Types.INTEGER, "ID", Integer.class, "id"));
+ storeTypes[1].setKeyFields(new JdbcTypeField(Types.INTEGER, "ID", Integer.class, "id"));
storeTypes[1].setValueType("org.apache.ignite.cache.store.jdbc.model.Person");
storeTypes[1].setValueFields(
- new CacheJdbcPojoStoreTypeField(Types.INTEGER, "ID", Integer.class, "id"),
- new CacheJdbcPojoStoreTypeField(Types.INTEGER, "ORG_ID", Integer.class, "orgId"),
- new CacheJdbcPojoStoreTypeField(Types.VARCHAR, "NAME", String.class, "name"));
+ new JdbcTypeField(Types.INTEGER, "ID", Integer.class, "id"),
+ new JdbcTypeField(Types.INTEGER, "ORG_ID", Integer.class, "orgId"),
+ new JdbcTypeField(Types.VARCHAR, "NAME", String.class, "name"));
- storeTypes[2] = new CacheJdbcPojoStoreType();
+ storeTypes[2] = new JdbcType();
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(Types.INTEGER, "ID", int.class, "id"),
- new CacheJdbcPojoStoreTypeField(Types.INTEGER, "ORG_ID", int.class, "orgId"),
- new CacheJdbcPojoStoreTypeField(Types.INTEGER, "CITY_ID", int.class, "cityId"));
+ new JdbcTypeField(Types.INTEGER, "ID", int.class, "id"),
+ new JdbcTypeField(Types.INTEGER, "ORG_ID", int.class, "orgId"),
+ new JdbcTypeField(Types.INTEGER, "CITY_ID", int.class, "cityId"));
storeTypes[2].setValueType("org.apache.ignite.cache.store.jdbc.model.Person");
storeTypes[2].setValueFields(
- new CacheJdbcPojoStoreTypeField(Types.INTEGER, "ID", Integer.class, "id"),
- new CacheJdbcPojoStoreTypeField(Types.INTEGER, "ORG_ID", Integer.class, "orgId"),
- new CacheJdbcPojoStoreTypeField(Types.VARCHAR, "NAME", String.class, "name"),
- new CacheJdbcPojoStoreTypeField(Types.INTEGER, "SALARY", Integer.class, "salary"));
+ new JdbcTypeField(Types.INTEGER, "ID", Integer.class, "id"),
+ new JdbcTypeField(Types.INTEGER, "ORG_ID", Integer.class, "orgId"),
+ new JdbcTypeField(Types.VARCHAR, "NAME", String.class, "name"),
+ new JdbcTypeField(Types.INTEGER, "SALARY", Integer.class, "salary"));
- storeTypes[3] = new CacheJdbcPojoStoreType();
+ storeTypes[3] = new JdbcType();
storeTypes[3].setDatabaseSchema("PUBLIC");
storeTypes[3].setDatabaseTable("TIMESTAMP_ENTRIES");
storeTypes[3].setKeyType("java.sql.Timestamp");
- storeTypes[3].setKeyFields(new CacheJdbcPojoStoreTypeField(Types.TIMESTAMP, "KEY", Timestamp.class, null));
+ storeTypes[3].setKeyFields(new JdbcTypeField(Types.TIMESTAMP, "KEY", Timestamp.class, null));
storeTypes[3].setValueType("java.lang.Integer");
- storeTypes[3].setValueFields(new CacheJdbcPojoStoreTypeField(Types.INTEGER, "VAL", Integer.class, null));
+ storeTypes[3].setValueFields(new JdbcTypeField(Types.INTEGER, "VAL", Integer.class, null));
- storeTypes[4] = new CacheJdbcPojoStoreType();
+ storeTypes[4] = new JdbcType();
storeTypes[4].setDatabaseSchema("PUBLIC");
storeTypes[4].setDatabaseTable("STRING_ENTRIES");
storeTypes[4].setKeyType("java.lang.String");
- storeTypes[4].setKeyFields(new CacheJdbcPojoStoreTypeField(Types.VARCHAR, "KEY", String.class, null));
+ storeTypes[4].setKeyFields(new JdbcTypeField(Types.VARCHAR, "KEY", String.class, null));
storeTypes[4].setValueType("java.lang.String");
- storeTypes[4].setValueFields(new CacheJdbcPojoStoreTypeField(Types.VARCHAR, "VAL", Integer.class, null));
+ storeTypes[4].setValueFields(new JdbcTypeField(Types.VARCHAR, "VAL", Integer.class, null));
- storeTypes[5] = new CacheJdbcPojoStoreType();
+ storeTypes[5] = new JdbcType();
storeTypes[5].setDatabaseSchema("PUBLIC");
storeTypes[5].setDatabaseTable("UUID_ENTRIES");
storeTypes[5].setKeyType("java.util.UUID");
- storeTypes[5].setKeyFields(new CacheJdbcPojoStoreTypeField(Types.BINARY, "KEY", UUID.class, null));
+ storeTypes[5].setKeyFields(new JdbcTypeField(Types.BINARY, "KEY", UUID.class, null));
storeTypes[5].setValueType("java.util.UUID");
- storeTypes[5].setValueFields(new CacheJdbcPojoStoreTypeField(Types.BINARY, "VAL", UUID.class, null));
+ storeTypes[5].setValueFields(new JdbcTypeField(Types.BINARY, "VAL", UUID.class, null));
storeCfg.setTypes(storeTypes);
http://git-wip-us.apache.org/repos/asf/ignite/blob/2961aaa7/modules/core/src/test/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPortableStoreSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPortableStoreSelfTest.java b/modules/core/src/test/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPortableStoreSelfTest.java
index 4b533b2..299b537 100644
--- a/modules/core/src/test/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPortableStoreSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPortableStoreSelfTest.java
@@ -17,29 +17,11 @@
package org.apache.ignite.cache.store.jdbc;
-import java.sql.Connection;
-import java.sql.DriverManager;
-import java.sql.PreparedStatement;
-import java.sql.SQLException;
-import java.sql.Statement;
import java.sql.Types;
import java.util.ArrayList;
import java.util.Collection;
-import org.apache.ignite.IgniteCache;
-import org.apache.ignite.cache.store.jdbc.dialect.H2Dialect;
-import org.apache.ignite.configuration.CacheConfiguration;
-import org.apache.ignite.configuration.IgniteConfiguration;
-import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.marshaller.Marshaller;
import org.apache.ignite.marshaller.portable.PortableMarshaller;
-import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
-import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
-import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
-import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
-import org.h2.jdbcx.JdbcConnectionPool;
-
-import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC;
-import static org.apache.ignite.cache.CacheMode.PARTITIONED;
/**
* Class for {@code PojoCacheStore} tests.
@@ -61,32 +43,32 @@ public class CacheJdbcPortableStoreSelfTest extends CacheJdbcStoreAbstractSelfTe
}
/** {@inheritDoc} */
- @Override protected CacheJdbcPojoStoreType[] storeTypes() {
- CacheJdbcPojoStoreType[] storeTypes = new CacheJdbcPojoStoreType[2];
+ @Override protected JdbcType[] storeTypes() {
+ JdbcType[] storeTypes = new JdbcType[2];
- storeTypes[0] = new CacheJdbcPojoStoreType();
+ storeTypes[0] = new JdbcType();
storeTypes[0].setKeepSerialized(true);
storeTypes[0].setDatabaseSchema("PUBLIC");
storeTypes[0].setDatabaseTable("ORGANIZATION");
storeTypes[0].setKeyType("org.apache.ignite.cache.store.jdbc.model.OrganizationKey");
- storeTypes[0].setKeyFields(new CacheJdbcPojoStoreTypeField(Types.INTEGER, "ID", Integer.class, "id"));
+ storeTypes[0].setKeyFields(new JdbcTypeField(Types.INTEGER, "ID", Integer.class, "id"));
storeTypes[0].setValueType("org.apache.ignite.cache.store.jdbc.model.Organization");
storeTypes[0].setValueFields(
- new CacheJdbcPojoStoreTypeField(Types.INTEGER, "ID", Integer.class, "id"),
- new CacheJdbcPojoStoreTypeField(Types.VARCHAR, "NAME", String.class, "name"),
- new CacheJdbcPojoStoreTypeField(Types.VARCHAR, "CITY", String.class, "city"));
+ new JdbcTypeField(Types.INTEGER, "ID", Integer.class, "id"),
+ new JdbcTypeField(Types.VARCHAR, "NAME", String.class, "name"),
+ new JdbcTypeField(Types.VARCHAR, "CITY", String.class, "city"));
- storeTypes[1] = new CacheJdbcPojoStoreType();
+ storeTypes[1] = new JdbcType();
storeTypes[1].setKeepSerialized(true);
storeTypes[1].setDatabaseSchema("PUBLIC");
storeTypes[1].setDatabaseTable("PERSON");
storeTypes[1].setKeyType("org.apache.ignite.cache.store.jdbc.model.PersonKey");
- storeTypes[1].setKeyFields(new CacheJdbcPojoStoreTypeField(Types.INTEGER, "ID", Integer.class, "id"));
+ storeTypes[1].setKeyFields(new JdbcTypeField(Types.INTEGER, "ID", Integer.class, "id"));
storeTypes[1].setValueType("org.apache.ignite.cache.store.jdbc.model.Person");
storeTypes[1].setValueFields(
- new CacheJdbcPojoStoreTypeField(Types.INTEGER, "ID", Integer.class, "id"),
- new CacheJdbcPojoStoreTypeField(Types.INTEGER, "ORG_ID", Integer.class, "orgId"),
- new CacheJdbcPojoStoreTypeField(Types.VARCHAR, "NAME", String.class, "name"));
+ new JdbcTypeField(Types.INTEGER, "ID", Integer.class, "id"),
+ new JdbcTypeField(Types.INTEGER, "ORG_ID", Integer.class, "orgId"),
+ new JdbcTypeField(Types.VARCHAR, "NAME", String.class, "name"));
return storeTypes;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2961aaa7/modules/core/src/test/java/org/apache/ignite/cache/store/jdbc/CacheJdbcStoreAbstractMultithreadedSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/cache/store/jdbc/CacheJdbcStoreAbstractMultithreadedSelfTest.java b/modules/core/src/test/java/org/apache/ignite/cache/store/jdbc/CacheJdbcStoreAbstractMultithreadedSelfTest.java
index 42cc4c9..4c4bd58 100644
--- a/modules/core/src/test/java/org/apache/ignite/cache/store/jdbc/CacheJdbcStoreAbstractMultithreadedSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/cache/store/jdbc/CacheJdbcStoreAbstractMultithreadedSelfTest.java
@@ -60,7 +60,7 @@ import static org.apache.ignite.testframework.GridTestUtils.runMultiThreadedAsyn
/**
*
*/
-public abstract class CacheJdbcStoreAbstractMultithreadedSelfTest<T extends CacheJdbcPojoStore>
+public abstract class CacheJdbcStoreAbstractMultithreadedSelfTest<T extends CacheAbstractJdbcStore>
extends GridCommonAbstractTest {
/** Default config with mapping. */
private static final String DFLT_MAPPING_CONFIG = "modules/core/src/test/config/store/jdbc/ignite-type-metadata.xml";
@@ -78,7 +78,7 @@ public abstract class CacheJdbcStoreAbstractMultithreadedSelfTest<T extends Cach
private static final int BATCH_CNT = 2000;
/** Cache store. */
- protected static CacheJdbcPojoStore store;
+ protected static CacheAbstractJdbcStore store;
/** {@inheritDoc} */
@Override protected void beforeTestsStarted() throws Exception {
http://git-wip-us.apache.org/repos/asf/ignite/blob/2961aaa7/modules/core/src/test/java/org/apache/ignite/cache/store/jdbc/CacheJdbcStoreAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/cache/store/jdbc/CacheJdbcStoreAbstractSelfTest.java b/modules/core/src/test/java/org/apache/ignite/cache/store/jdbc/CacheJdbcStoreAbstractSelfTest.java
index c083e93..2b07b0a 100644
--- a/modules/core/src/test/java/org/apache/ignite/cache/store/jdbc/CacheJdbcStoreAbstractSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/cache/store/jdbc/CacheJdbcStoreAbstractSelfTest.java
@@ -22,7 +22,6 @@ import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Statement;
-import java.sql.Types;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.cache.store.jdbc.dialect.H2Dialect;
import org.apache.ignite.configuration.CacheConfiguration;
@@ -30,7 +29,6 @@ import org.apache.ignite.configuration.ConnectorConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.marshaller.Marshaller;
-import org.apache.ignite.marshaller.portable.PortableMarshaller;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
@@ -110,6 +108,9 @@ public abstract class CacheJdbcStoreAbstractSelfTest extends GridCommonAbstractT
return cfg;
}
+ /**
+ * @return Marshaller to be used in test.
+ */
protected abstract Marshaller marshaller();
/** */
@@ -123,7 +124,10 @@ public abstract class CacheJdbcStoreAbstractSelfTest extends GridCommonAbstractT
return storeCfg;
}
- protected abstract CacheJdbcPojoStoreType[] storeTypes();
+ /**
+ * @return Types to be used in test.
+ */
+ protected abstract JdbcType[] storeTypes();
/** */
protected CacheConfiguration cacheConfiguration() throws Exception {
@@ -133,9 +137,8 @@ public abstract class CacheJdbcStoreAbstractSelfTest extends GridCommonAbstractT
cc.setAtomicityMode(ATOMIC);
cc.setSwapEnabled(false);
cc.setWriteBehindEnabled(false);
- cc.setNearConfiguration(null);
- CacheJdbcPojoStoreFactory<Object, Object> storeFactory = new CacheJdbcPojoStoreFactory<Object, Object>();
+ CacheJdbcPojoStoreFactory<Object, Object> storeFactory = new CacheJdbcPojoStoreFactory<>();
storeFactory.setConfiguration(storeConfiguration());
storeFactory.setDataSource(JdbcConnectionPool.create(DFLT_CONN_URL, "sa", "")); // H2 DataSource
@@ -147,6 +150,11 @@ public abstract class CacheJdbcStoreAbstractSelfTest extends GridCommonAbstractT
return cc;
}
+ /**
+ * Fill in-memory database with sample data.
+ *
+ * @throws SQLException In case of filling database with sample data failed.
+ */
protected void fillSampleDatabase() throws SQLException {
Connection conn = getConnection();
[2/4] ignite git commit: IGNITE-1753 Wip reworking code.
Posted by ak...@apache.org.
http://git-wip-us.apache.org/repos/asf/ignite/blob/2961aaa7/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 f8abd16..1724fa0 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
@@ -19,127 +19,30 @@ package org.apache.ignite.cache.store.jdbc;
import java.lang.reflect.Constructor;
import java.lang.reflect.Method;
-import java.nio.ByteBuffer;
-import java.sql.BatchUpdateException;
-import java.sql.Connection;
-import java.sql.PreparedStatement;
import java.sql.ResultSet;
-import java.sql.ResultSetMetaData;
import java.sql.SQLException;
-import java.sql.Statement;
-import java.sql.Types;
-import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
-import java.util.List;
import java.util.Map;
-import java.util.UUID;
-import java.util.concurrent.Callable;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
-import javax.cache.Cache;
import javax.cache.CacheException;
import javax.cache.integration.CacheLoaderException;
-import javax.cache.integration.CacheWriterException;
-import javax.sql.DataSource;
import org.apache.ignite.Ignite;
-import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.IgniteException;
-import org.apache.ignite.IgniteLogger;
import org.apache.ignite.IgnitePortables;
-import org.apache.ignite.cache.CacheTypeFieldMetadata;
-import org.apache.ignite.cache.CacheTypeMetadata;
import org.apache.ignite.cache.store.CacheStore;
-import org.apache.ignite.cache.store.CacheStoreSession;
-import org.apache.ignite.cache.store.jdbc.dialect.BasicJdbcDialect;
-import org.apache.ignite.cache.store.jdbc.dialect.DB2Dialect;
-import org.apache.ignite.cache.store.jdbc.dialect.H2Dialect;
-import org.apache.ignite.cache.store.jdbc.dialect.JdbcDialect;
-import org.apache.ignite.cache.store.jdbc.dialect.MySQLDialect;
-import org.apache.ignite.cache.store.jdbc.dialect.OracleDialect;
-import org.apache.ignite.cache.store.jdbc.dialect.SQLServerDialect;
import org.apache.ignite.configuration.CacheConfiguration;
-import org.apache.ignite.internal.util.tostring.GridToStringExclude;
-import org.apache.ignite.internal.util.typedef.C1;
-import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.U;
-import org.apache.ignite.lang.IgniteBiInClosure;
-import org.apache.ignite.lang.IgnitePredicate;
-import org.apache.ignite.lifecycle.LifecycleAware;
import org.apache.ignite.portable.PortableBuilder;
import org.apache.ignite.portable.PortableObject;
-import org.apache.ignite.resources.CacheStoreSessionResource;
-import org.apache.ignite.resources.IgniteInstanceResource;
-import org.apache.ignite.resources.LoggerResource;
-import org.apache.ignite.transactions.Transaction;
import org.jetbrains.annotations.Nullable;
-import static java.sql.Statement.EXECUTE_FAILED;
-import static java.sql.Statement.SUCCESS_NO_INFO;
-import static org.apache.ignite.cache.store.jdbc.CacheJdbcPojoStoreConfiguration.DFLT_BATCH_SIZE;
-import static org.apache.ignite.cache.store.jdbc.CacheJdbcPojoStoreConfiguration.DFLT_WRITE_ATTEMPTS;
-import static org.apache.ignite.cache.store.jdbc.CacheJdbcPojoStoreConfiguration.DFLT_PARALLEL_LOAD_CACHE_MINIMUM_THRESHOLD;
-
/**
* Implementation of {@link CacheStore} backed by JDBC and POJO via reflection.
*
* This implementation stores objects in underlying database using java beans mapping description via reflection. <p>
* Use {@link CacheJdbcPojoStoreFactory} factory to pass {@link CacheJdbcPojoStore} to {@link CacheConfiguration}.
*/
-public class CacheJdbcPojoStore<K, V> implements CacheStore<K, V>, LifecycleAware {
- /** Connection attribute property name. */
- private static final String ATTR_CONN_PROP = "JDBC_STORE_CONNECTION";
-
- /** Auto-injected store session. */
- @CacheStoreSessionResource
- private CacheStoreSession ses;
-
- /** Auto injected ignite instance. */
- @IgniteInstanceResource
- private Ignite ignite;
-
- /** Auto-injected logger instance. */
- @LoggerResource
- private IgniteLogger log;
-
- /** Lock for metadata cache. */
- @GridToStringExclude
- private final Lock cacheMappingsLock = new ReentrantLock();
-
- /** Data source. */
- private DataSource dataSrc;
-
- /** Portables. */
- private IgnitePortables portables;
-
- /** Cache with entry mapping description. (cache name, (key id, mapping description)). */
- private volatile Map<String, Map<Object, EntryMapping>> cacheMappings = Collections.emptyMap();
-
- /** Maximum batch size for writeAll and deleteAll operations. */
- private int batchSz = DFLT_BATCH_SIZE;
-
- /** Database dialect. */
- private JdbcDialect dialect;
-
- /** Max workers thread count. These threads are responsible for load cache. */
- private int maxPoolSz = Runtime.getRuntime().availableProcessors();
-
- /** Maximum write attempts in case of database error. */
- private int maxWrtAttempts = DFLT_WRITE_ATTEMPTS;
-
- /** Parallel load cache minimum threshold. If {@code 0} then load sequentially. */
- private int parallelLoadCacheMinThreshold = DFLT_PARALLEL_LOAD_CACHE_MINIMUM_THRESHOLD;
-
- /** Types that store could process. */
- private CacheJdbcPojoStoreType[] types;
-
- /** Map for quick check whether type is POJO or Portable. */
- private volatile Map<String, Map<String, Boolean>> keepSerializedTypes = new HashMap<>();
-
+public class CacheJdbcPojoStore<K, V> extends CacheAbstractJdbcStore<K, V> {
/** POJO methods cache. */
private volatile Map<String, Map<String, PojoMethodsCache>> pojoMethods = Collections.emptyMap();
@@ -147,28 +50,6 @@ public class CacheJdbcPojoStore<K, V> implements CacheStore<K, V>, LifecycleAwar
private volatile Map<String, Map<String, Integer>> portableTypeIds = Collections.emptyMap();
/**
- * Checks for POJO/portable format.
- *
- * @param cacheName Cache name to get types settings.
- * @param typeName Type name to check for POJO/portable format.
- * @return {@code true} If portable format configured.
- * @throws CacheException In case of error.
- */
- private boolean isKeepSerialized(String cacheName, String typeName) {
- Map<String, Boolean> cacheTypes = keepSerializedTypes.get(cacheName);
-
- if (cacheTypes == null)
- throw new CacheException("Failed to find types metadata for cache: " + cacheName);
-
- Boolean keepSerialized = cacheTypes.get(typeName);
-
- if (keepSerialized == null)
- throw new CacheException("Failed to find type metadata for type: " + typeName);
-
- return keepSerialized;
- }
-
- /**
* Get field value from object for use as query parameter.
*
* @param cacheName Cache name.
@@ -178,7 +59,7 @@ public class CacheJdbcPojoStore<K, V> implements CacheStore<K, V>, LifecycleAwar
* @return Field value from object.
* @throws CacheException in case of error.
*/
- @Nullable private Object extractParameter(@Nullable String cacheName, String typeName, String fieldName,
+ @Nullable protected Object extractParameter(@Nullable String cacheName, String typeName, String fieldName,
Object obj) throws CacheException {
return isKeepSerialized(cacheName, typeName)
? extractPortableParameter(fieldName, obj)
@@ -255,8 +136,8 @@ public class CacheJdbcPojoStore<K, V> implements CacheStore<K, V>, LifecycleAwar
* @return Constructed object.
* @throws CacheLoaderException If failed to construct cache object.
*/
- private <R> R buildObject(@Nullable String cacheName, String typeName,
- CacheJdbcPojoStoreTypeField[] fields, Map<String, Integer> loadColIdxs, ResultSet rs)
+ @Override protected <R> R buildObject(@Nullable String cacheName, String typeName,
+ JdbcTypeField[] fields, Map<String, Integer> loadColIdxs, ResultSet rs)
throws CacheLoaderException {
return (R)(isKeepSerialized(cacheName, typeName)
? buildPortableObject(cacheName, typeName, fields, loadColIdxs, rs)
@@ -275,7 +156,7 @@ public class CacheJdbcPojoStore<K, V> implements CacheStore<K, V>, LifecycleAwar
* @throws CacheLoaderException If failed to construct POJO.
*/
private Object buildPojoObject(@Nullable String cacheName, String typeName,
- CacheJdbcPojoStoreTypeField[] fields, Map<String, Integer> loadColIdxs, ResultSet rs)
+ JdbcTypeField[] fields, Map<String, Integer> loadColIdxs, ResultSet rs)
throws CacheLoaderException {
Map<String, PojoMethodsCache> cacheMethods = pojoMethods.get(cacheName);
@@ -290,14 +171,14 @@ public class CacheJdbcPojoStore<K, V> implements CacheStore<K, V>, LifecycleAwar
try {
if (mc.simple) {
- CacheJdbcPojoStoreTypeField field = fields[0];
+ JdbcTypeField field = fields[0];
return getColumnValue(rs, loadColIdxs.get(field.getDatabaseFieldName()), mc.cls);
}
Object obj = mc.ctor.newInstance();
- for (CacheJdbcPojoStoreTypeField field : fields) {
+ for (JdbcTypeField field : fields) {
String fldJavaName = field.getJavaFieldName();
Method setter = mc.setters.get(fldJavaName);
@@ -340,7 +221,7 @@ public class CacheJdbcPojoStore<K, V> implements CacheStore<K, V>, LifecycleAwar
* @return Constructed portable object.
* @throws CacheLoaderException If failed to construct portable object.
*/
- protected PortableObject buildPortableObject(String cacheName, String typeName, CacheJdbcPojoStoreTypeField[] fields,
+ protected PortableObject buildPortableObject(String cacheName, String typeName, JdbcTypeField[] fields,
Map<String, Integer> loadColIdxs, ResultSet rs) throws CacheException {
Map<String, Integer> cacheTypeIds = portableTypeIds.get(cacheName);
@@ -352,21 +233,28 @@ public class CacheJdbcPojoStore<K, V> implements CacheStore<K, V>, LifecycleAwar
if (typeId == null)
throw new CacheLoaderException("Failed to find portable type ID for type: " + typeName);
- PortableBuilder builder = portables.builder(typeId);
+ PortableBuilder builder = ignite.portables().builder(typeId);
if (builder == null)
throw new CacheException("Failed to find portable builder for type: " + typeName);
try {
- for (CacheJdbcPojoStoreTypeField field : fields) {
+ int hashCode = 1;
+
+ for (JdbcTypeField field : fields) {
Class<?> type = field.getJavaFieldType();
Integer colIdx = loadColIdxs.get(field.getDatabaseFieldName());
- builder.setField(field.getJavaFieldName(), getColumnValue(rs, colIdx, type));
+ Object colVal = getColumnValue(rs, colIdx, type);
+
+ if (colVal != null)
+ hashCode = 31 * hashCode + colVal.hashCode();
+
+ builder.setField(field.getJavaFieldName(), colVal);
}
- return builder.build();
+ return builder.hashCode(hashCode).build();
}
catch (SQLException e) {
throw new CacheException("Failed to read portable object", e);
@@ -380,7 +268,7 @@ public class CacheJdbcPojoStore<K, V> implements CacheStore<K, V>, LifecycleAwar
* @return Type ID.
* @throws CacheException If failed to calculate type ID for given object.
*/
- private Object typeIdForObject(Object obj) throws CacheException {
+ @Override protected Object typeIdForObject(Object obj) throws CacheException {
if (obj instanceof PortableObject)
return ((PortableObject)obj).typeId();
@@ -395,9 +283,9 @@ public class CacheJdbcPojoStore<K, V> implements CacheStore<K, V>, LifecycleAwar
* @return Type ID.
* @throws CacheException If failed to get type ID for given type name.
*/
- private Object typeIdForTypeName(boolean keepSerialized, String typeName) throws CacheException {
+ @Override protected Object typeIdForTypeName(boolean keepSerialized, String typeName) throws CacheException {
if (keepSerialized)
- return portables.typeId(typeName);
+ return ignite.portables().typeId(typeName);
try {
return Class.forName(typeName);
@@ -408,17 +296,31 @@ public class CacheJdbcPojoStore<K, V> implements CacheStore<K, V>, LifecycleAwar
}
/**
+ * Prepare internal store specific builders for provided types metadata.
+ *
+ * @param cacheName Cache name to prepare builders for.
+ * @param types Collection of types.
+ * @throws CacheException If failed to prepare internal builders for types.
+ */
+ @Override protected void prepareBuilders(@Nullable String cacheName, Collection<JdbcType> types)
+ throws CacheException {
+ preparePojoBuilders(cacheName, types);
+ preparePortableBuilders(cacheName, types);
+ }
+
+
+ /**
* Prepare builders for POJOs via reflection (getters and setters).
*
* @param cacheName Cache name to prepare builders for.
* @param types Collection of types.
* @throws CacheException If failed to prepare internal builders for types.
*/
- private void preparePojoBuilders(@Nullable String cacheName, Collection<CacheJdbcPojoStoreType> types)
+ private void preparePojoBuilders(@Nullable String cacheName, Collection<JdbcType> types)
throws CacheException {
Map<String, PojoMethodsCache> typeMethods = U.newHashMap(types.size() * 2);
- for (CacheJdbcPojoStoreType type : types) {
+ for (JdbcType type : types) {
if (!type.isKeepSerialized()) {
String keyType = type.getKeyType();
@@ -450,11 +352,11 @@ 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.
*/
- private void preparePortableBuilders(@Nullable String cacheName, Collection<CacheJdbcPojoStoreType> types)
+ private void preparePortableBuilders(@Nullable String cacheName, Collection<JdbcType> types)
throws CacheException {
Map<String, Integer> typeIds = U.newHashMap(types.size() * 2);
- for (CacheJdbcPojoStoreType type : types) {
+ for (JdbcType type : types) {
if (type.isKeepSerialized()) {
Ignite ignite = ignite();
@@ -478,1438 +380,76 @@ public class CacheJdbcPojoStore<K, V> implements CacheStore<K, V>, LifecycleAwar
}
/**
- * Perform dialect resolution.
- *
- * @return The resolved dialect.
- * @throws CacheException Indicates problems accessing the metadata.
- */
- private JdbcDialect resolveDialect() throws CacheException {
- Connection conn = null;
-
- String dbProductName = null;
-
- try {
- conn = openConnection(false);
-
- dbProductName = conn.getMetaData().getDatabaseProductName();
- }
- catch (SQLException e) {
- throw new CacheException("Failed access to metadata for detect database dialect.", e);
- }
- finally {
- U.closeQuiet(conn);
- }
-
- if ("H2".equals(dbProductName))
- return new H2Dialect();
-
- if ("MySQL".equals(dbProductName))
- return new MySQLDialect();
-
- if (dbProductName.startsWith("Microsoft SQL Server"))
- return new SQLServerDialect();
-
- if ("Oracle".equals(dbProductName))
- return new OracleDialect();
-
- if (dbProductName.startsWith("DB2/"))
- return new DB2Dialect();
-
- U.warn(log, "Failed to resolve dialect (BasicJdbcDialect will be used): " + dbProductName);
-
- return new BasicJdbcDialect();
- }
-
- /** {@inheritDoc} */
- @Override public void start() throws IgniteException {
- if (dataSrc == null)
- throw new IgniteException("Failed to initialize cache store (data source is not provided).");
-
- if (dialect == null) {
- dialect = resolveDialect();
-
- if (log.isDebugEnabled() && dialect.getClass() != BasicJdbcDialect.class)
- log.debug("Resolved database dialect: " + U.getSimpleName(dialect.getClass()));
- }
-
- portables = ignite.portables();
- }
-
- /** {@inheritDoc} */
- @Override public void stop() throws IgniteException {
- // No-op.
- }
-
- /**
- * Gets connection from a pool.
- *
- * @param autocommit {@code true} If connection should use autocommit mode.
- * @return Pooled connection.
- * @throws SQLException In case of error.
- */
- protected Connection openConnection(boolean autocommit) throws SQLException {
- Connection conn = dataSrc.getConnection();
-
- conn.setAutoCommit(autocommit);
-
- return conn;
- }
-
- /**
- * @return Connection.
- * @throws SQLException In case of error.
- */
- private Connection connection() throws SQLException {
- CacheStoreSession ses = session();
-
- if (ses.transaction() != null) {
- Map<String, Connection> prop = ses.properties();
-
- Connection conn = prop.get(ATTR_CONN_PROP);
-
- if (conn == null) {
- conn = openConnection(false);
-
- // Store connection in session to used it for other operations in the same session.
- prop.put(ATTR_CONN_PROP, conn);
- }
-
- return conn;
- }
- // Transaction can be null in case of simple load operation.
- else
- return openConnection(true);
- }
-
- /**
- * Closes connection.
- *
- * @param conn Connection to close.
+ * POJO methods cache.
*/
- private void closeConnection(@Nullable Connection conn) {
- CacheStoreSession ses = session();
+ private static class PojoMethodsCache {
+ /** POJO class. */
+ private final Class<?> cls;
- // Close connection right away if there is no transaction.
- if (ses.transaction() == null)
- U.closeQuiet(conn);
- }
+ /** Constructor for POJO object. */
+ private Constructor ctor;
- /**
- * Closes allocated resources depending on transaction status.
- *
- * @param conn Allocated connection.
- * @param st Created statement,
- */
- private void end(@Nullable Connection conn, @Nullable Statement st) {
- U.closeQuiet(st);
+ /** {@code true} if object is a simple type. */
+ private final boolean simple;
- closeConnection(conn);
- }
+ /** Cached setters for POJO object. */
+ private Map<String, Method> getters;
- /** {@inheritDoc} */
- @Override public void sessionEnd(boolean commit) throws CacheWriterException {
- CacheStoreSession ses = session();
+ /** Cached getters for POJO object. */
+ private Map<String, Method> setters;
- Transaction tx = ses.transaction();
+ /**
+ * POJO methods cache.
+ *
+ * @param clsName Class name.
+ * @param fields Fields.
+ * @throws CacheException If failed to construct type cache.
+ */
+ public PojoMethodsCache(String clsName, JdbcTypeField[] fields) throws CacheException {
+ try {
+ cls = Class.forName(clsName);
- if (tx != null) {
- Connection conn = ses.<String, Connection>properties().remove(ATTR_CONN_PROP);
+ if (simple = simpleType(cls))
+ return;
- assert conn != null;
+ ctor = cls.getDeclaredConstructor();
- try {
- if (commit)
- conn.commit();
- else
- conn.rollback();
- }
- catch (SQLException e) {
- throw new CacheWriterException(
- "Failed to end transaction [xid=" + tx.xid() + ", commit=" + commit + ']', e);
+ if (!ctor.isAccessible())
+ ctor.setAccessible(true);
}
- finally {
- U.closeQuiet(conn);
+ catch (ClassNotFoundException e) {
+ throw new CacheException("Failed to find class: " + clsName, e);
}
-
- if (log.isDebugEnabled())
- log.debug("Transaction ended [xid=" + tx.xid() + ", commit=" + commit + ']');
- }
- }
-
- /**
- * Retrieves the value of the designated column in the current row of this <code>ResultSet</code> object and will
- * convert to the requested Java data type.
- *
- * @param rs Result set.
- * @param colIdx Column index in result set.
- * @param type Class representing the Java data type to convert the designated column to.
- * @return Value in column.
- * @throws SQLException If a database access error occurs or this method is called.
- */
- private Object getColumnValue(ResultSet rs, int colIdx, Class<?> type) throws SQLException {
- Object val = rs.getObject(colIdx);
-
- if (val == null)
- return null;
-
- if (type == int.class)
- return rs.getInt(colIdx);
-
- if (type == long.class)
- return rs.getLong(colIdx);
-
- if (type == double.class)
- return rs.getDouble(colIdx);
-
- if (type == boolean.class || type == Boolean.class)
- return rs.getBoolean(colIdx);
-
- if (type == byte.class)
- return rs.getByte(colIdx);
-
- if (type == short.class)
- return rs.getShort(colIdx);
-
- if (type == float.class)
- return rs.getFloat(colIdx);
-
- if (type == Integer.class || type == Long.class || type == Double.class ||
- type == Byte.class || type == Short.class || type == Float.class) {
- Number num = (Number)val;
-
- if (type == Integer.class)
- return num.intValue();
- else if (type == Long.class)
- return num.longValue();
- else if (type == Double.class)
- return num.doubleValue();
- else if (type == Byte.class)
- return num.byteValue();
- else if (type == Short.class)
- return num.shortValue();
- else if (type == Float.class)
- return num.floatValue();
- }
-
- if (type == UUID.class) {
- if (val instanceof UUID)
- return val;
-
- if (val instanceof byte[]) {
- ByteBuffer bb = ByteBuffer.wrap((byte[])val);
-
- long most = bb.getLong();
- long least = bb.getLong();
-
- return new UUID(most, least);
+ catch (NoSuchMethodException e) {
+ throw new CacheException("Failed to find default constructor for class: " + clsName, e);
}
- if (val instanceof String)
- return UUID.fromString((String)val);
- }
-
- return val;
- }
+ setters = U.newHashMap(fields.length);
- /**
- * Construct load cache from range.
- *
- * @param em Type mapping description.
- * @param clo Closure that will be applied to loaded values.
- * @param lowerBound Lower bound for range.
- * @param upperBound Upper bound for range.
- * @return Callable for pool submit.
- */
- private Callable<Void> loadCacheRange(final EntryMapping em, final IgniteBiInClosure<K, V> clo,
- @Nullable final Object[] lowerBound, @Nullable final Object[] upperBound) {
- return new Callable<Void>() {
- @Override public Void call() throws Exception {
- Connection conn = null;
+ getters = U.newHashMap(fields.length);
- PreparedStatement stmt = null;
+ for (JdbcTypeField field : fields) {
+ String prop = capitalFirst(field.getJavaFieldName());
try {
- conn = openConnection(true);
-
- stmt = conn.prepareStatement(lowerBound == null && upperBound == null
- ? em.loadCacheQry
- : em.loadCacheRangeQuery(lowerBound != null, upperBound != null));
-
- int ix = 1;
-
- if (lowerBound != null)
- for (int i = lowerBound.length; i > 0; i--)
- for (int j = 0; j < i; j++)
- stmt.setObject(ix++, lowerBound[j]);
-
- if (upperBound != null)
- for (int i = upperBound.length; i > 0; i--)
- for (int j = 0; j < i; j++)
- stmt.setObject(ix++, upperBound[j]);
-
- ResultSet rs = stmt.executeQuery();
-
- long t = System.currentTimeMillis();
-
- while (rs.next()) {
- K key = buildObject(em.cacheName, em.keyType(), em.keyColumns(), em.loadColIdxs, rs);
- V val = buildObject(em.cacheName, em.valueType(), em.valueColumns(), em.loadColIdxs, rs);
-
- clo.apply(key, val);
- }
- }
- catch (SQLException e) {
- throw new IgniteCheckedException("Failed to load cache", e);
- }
- finally {
- U.closeQuiet(stmt);
-
- U.closeQuiet(conn);
+ getters.put(field.getJavaFieldName(), cls.getMethod("get" + prop));
}
-
- return null;
- }
- };
- }
-
- /**
- * Construct load cache in one select.
- *
- * @param m Type mapping description.
- * @param clo Closure for loaded values.
- * @return Callable for pool submit.
- */
- private Callable<Void> loadCacheFull(EntryMapping m, IgniteBiInClosure<K, V> clo) {
- return loadCacheRange(m, clo, null, null);
- }
-
- /**
- * Object is a simple type.
- *
- * @param cls Class.
- * @return {@code True} if object is a simple type.
- */
- 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));
- }
-
- /**
- * @param cacheName Cache name to check mapping for.
- * @param clsName Class name.
- * @param fields Fields descriptors.
- * @throws CacheException If failed to check type metadata.
- */
- private static void checkMapping(@Nullable String cacheName, String clsName,
- CacheJdbcPojoStoreTypeField[] fields) throws CacheException {
- try {
- Class<?> cls = Class.forName(clsName);
-
- if (simpleType(cls)) {
- if (fields.length != 1)
- throw new CacheException("More than one field for simple type [cache name=" + cacheName
- + ", type=" + clsName + " ]");
-
- CacheJdbcPojoStoreTypeField field = fields[0];
-
- if (field.getDatabaseFieldName() == null)
- throw new CacheException("Missing database name in mapping description [cache name=" + cacheName
- + ", type=" + clsName + " ]");
-
- field.setJavaFieldType(cls);
- }
- else
- for (CacheJdbcPojoStoreTypeField field : fields) {
- if (field.getDatabaseFieldName() == null)
- throw new CacheException("Missing database name in mapping description [cache name=" + cacheName
- + ", type=" + clsName + " ]");
-
- if (field.getJavaFieldName() == null)
- throw new CacheException("Missing field name in mapping description [cache name=" + cacheName
- + ", type=" + clsName + " ]");
-
- if (field.getJavaFieldType() == null)
- throw new CacheException("Missing field type in mapping description [cache name=" + cacheName
- + ", type=" + clsName + " ]");
+ catch (NoSuchMethodException ignored) {
+ try {
+ getters.put(field.getJavaFieldName(), cls.getMethod("is" + prop));
+ }
+ catch (NoSuchMethodException e) {
+ throw new CacheException("Failed to find getter in POJO class [clsName=" + clsName +
+ ", prop=" + field.getJavaFieldName() + "]", e);
+ }
}
- }
- catch (ClassNotFoundException e) {
- throw new CacheException("Failed to find class: " + clsName, e);
- }
- }
-
- /**
- * For backward compatibility translate old field type descriptors to new format.
- *
- * @param oldFields Fields in old format.
- * @return Fields in new format.
- */
- @Deprecated
- private CacheJdbcPojoStoreTypeField[] translateFields(Collection<CacheTypeFieldMetadata> oldFields) {
- CacheJdbcPojoStoreTypeField[] newFields = new CacheJdbcPojoStoreTypeField[oldFields.size()];
-
- int idx = 0;
-
- for (CacheTypeFieldMetadata oldField : oldFields) {
- newFields[idx] = new CacheJdbcPojoStoreTypeField(oldField.getDatabaseType(), oldField.getDatabaseName(),
- oldField.getJavaType(), oldField.getJavaName());
-
- idx++;
- }
-
- return newFields;
- }
-
- /**
- * @param cacheName Cache name to check mappings for.
- * @return Type mappings for specified cache name.
- * @throws CacheException If failed to initialize cache mappings.
- */
- private Map<Object, EntryMapping> cacheMappings(@Nullable String cacheName) throws CacheException {
- Map<Object, EntryMapping> entryMappings = cacheMappings.get(cacheName);
-
- if (entryMappings != null)
- return entryMappings;
-
- cacheMappingsLock.lock();
-
- try {
- entryMappings = cacheMappings.get(cacheName);
-
- if (entryMappings != null)
- return entryMappings;
-
- // If no types configured, check CacheTypeMetadata for backward compatibility.
- if (types == null) {
- CacheConfiguration ccfg = ignite().cache(cacheName).getConfiguration(CacheConfiguration.class);
-
- Collection<CacheTypeMetadata> oldTypes = ccfg.getTypeMetadata();
-
- types = new CacheJdbcPojoStoreType[oldTypes.size()];
-
- int idx = 0;
-
- for (CacheTypeMetadata oldType : oldTypes) {
- CacheJdbcPojoStoreType newType = new CacheJdbcPojoStoreType();
- newType.setCacheName(cacheName);
-
- newType.setDatabaseSchema(oldType.getDatabaseSchema());
- newType.setDatabaseTable(oldType.getDatabaseTable());
-
- newType.setKeyType(oldType.getKeyType());
- newType.setKeyFields(translateFields(oldType.getKeyFields()));
-
- newType.setValueType(oldType.getValueType());
- newType.setValueFields(translateFields(oldType.getValueFields()));
-
- types[idx] = newType;
-
- idx++;
+ try {
+ setters.put(field.getJavaFieldName(), cls.getMethod("set" + prop, field.getJavaFieldType()));
}
- }
-
- List<CacheJdbcPojoStoreType> cacheTypes = new ArrayList<>(types.length);
-
- for (CacheJdbcPojoStoreType type : types)
- if ((cacheName != null && cacheName.equals(type.getCacheName())) ||
- (cacheName == null && type.getCacheName() == null))
- cacheTypes.add(type);
-
- entryMappings = U.newHashMap(cacheTypes.size());
-
- if (!cacheTypes.isEmpty()) {
- Map<String, Boolean> tk = new HashMap<>(cacheTypes.size() * 2);
-
- for (CacheJdbcPojoStoreType type : cacheTypes) {
- boolean keepSerialized = type.isKeepSerialized();
-
- String keyType = type.getKeyType();
- String valType = type.getValueType();
-
- tk.put(keyType, keepSerialized);
- tk.put(valType, keepSerialized);
-
- Object keyTypeId = typeIdForTypeName(keepSerialized, keyType);
-
- if (entryMappings.containsKey(keyTypeId))
- throw new CacheException("Key type must be unique in type metadata [cache name=" + cacheName +
- ", key type=" + keyType + "]");
-
- if (!keepSerialized) {
- checkMapping(cacheName, keyType, type.getKeyFields());
- checkMapping(cacheName, valType, type.getValueFields());
- }
-
- entryMappings.put(keyTypeId, new EntryMapping(cacheName, dialect, type));
- }
-
- keepSerializedTypes.put(cacheName, tk);
-
- Map<String, Map<Object, EntryMapping>> mappings = new HashMap<>(cacheMappings);
- mappings.put(cacheName, entryMappings);
-
- preparePojoBuilders(cacheName, cacheTypes);
- preparePortableBuilders(cacheName, cacheTypes);
-
- cacheMappings = mappings;
- }
-
- return entryMappings;
- }
- finally {
- cacheMappingsLock.unlock();
- }
- }
-
- /**
- * @param cacheName Cache name.
- * @param keyTypeId Key type id.
- * @param key Key object.
- * @return Entry mapping.
- * @throws CacheException If mapping for key was not found.
- */
- private EntryMapping entryMapping(String cacheName, Object keyTypeId, Object key) throws CacheException {
- EntryMapping em = cacheMappings(cacheName).get(keyTypeId);
-
- if (em == null) {
- String maskedCacheName = U.maskName(cacheName);
-
- throw new CacheException("Failed to find mapping description [key=" + key +
- ", cache=" + maskedCacheName + "]. Please configure CacheJdbcPojoStoreType to associate '" + maskedCacheName + "' with JdbcPojoStore.");
- }
-
- return em;
- }
-
- /** {@inheritDoc} */
- @Override public void loadCache(final IgniteBiInClosure<K, V> clo, @Nullable Object... args)
- throws CacheLoaderException {
- ExecutorService pool = null;
-
- String cacheName = session().cacheName();
-
- try {
- pool = Executors.newFixedThreadPool(maxPoolSz);
-
- Collection<Future<?>> futs = new ArrayList<>();
-
- if (args != null && args.length > 0) {
- if (args.length % 2 != 0)
- throw new CacheLoaderException("Expected even number of arguments, but found: " + args.length);
-
- if (log.isDebugEnabled())
- log.debug("Start loading entries from db using user queries from arguments");
-
- for (int i = 0; i < args.length; i += 2) {
- String keyType = args[i].toString();
-
- String selQry = args[i + 1].toString();
-
- // We must build cache mappings first.
- cacheMappings(cacheName);
-
- EntryMapping em = entryMapping(cacheName, typeIdForTypeName(isKeepSerialized(cacheName, keyType),
- keyType), keyType);
-
- futs.add(pool.submit(new LoadCacheCustomQueryWorker<>(em, selQry, clo)));
- }
- }
- else {
- Collection<EntryMapping> entryMappings = cacheMappings(session().cacheName()).values();
-
- for (EntryMapping em : entryMappings) {
- if (parallelLoadCacheMinThreshold > 0) {
- if (log.isDebugEnabled())
- log.debug("Multithread loading entries from db [cache name=" + cacheName +
- ", key type=" + em.keyType() + " ]");
-
- Connection conn = null;
-
- try {
- conn = connection();
-
- PreparedStatement stmt = conn.prepareStatement(em.loadCacheSelRangeQry);
-
- stmt.setInt(1, parallelLoadCacheMinThreshold);
-
- ResultSet rs = stmt.executeQuery();
-
- if (rs.next()) {
- int keyCnt = em.keyCols.size();
-
- Object[] upperBound = new Object[keyCnt];
-
- for (int i = 0; i < keyCnt; i++)
- upperBound[i] = rs.getObject(i + 1);
-
- futs.add(pool.submit(loadCacheRange(em, clo, null, upperBound)));
-
- while (rs.next()) {
- Object[] lowerBound = upperBound;
-
- upperBound = new Object[keyCnt];
-
- for (int i = 0; i < keyCnt; i++)
- upperBound[i] = rs.getObject(i + 1);
-
- futs.add(pool.submit(loadCacheRange(em, clo, lowerBound, upperBound)));
- }
-
- futs.add(pool.submit(loadCacheRange(em, clo, upperBound, null)));
- }
- else
- futs.add(pool.submit(loadCacheFull(em, clo)));
- }
- catch (SQLException ignored) {
- futs.add(pool.submit(loadCacheFull(em, clo)));
- }
- finally {
- U.closeQuiet(conn);
- }
- }
- else {
- if (log.isDebugEnabled())
- log.debug("Single thread loading entries from db [cache name=" + cacheName +
- ", key type=" + em.keyType() + " ]");
-
- futs.add(pool.submit(loadCacheFull(em, clo)));
- }
- }
- }
-
- for (Future<?> fut : futs)
- U.get(fut);
-
- if (log.isDebugEnabled())
- log.debug("Cache loaded from db: " + cacheName);
- }
- catch (IgniteCheckedException e) {
- throw new CacheLoaderException("Failed to load cache: " + cacheName, e.getCause());
- }
- finally {
- U.shutdownNow(getClass(), pool, log);
- }
- }
-
- /** {@inheritDoc} */
- @Nullable @Override public V load(K key) throws CacheLoaderException {
- assert key != null;
-
- EntryMapping em = entryMapping(session().cacheName(), typeIdForObject(key), key);
-
- if (log.isDebugEnabled())
- log.debug("Load value from db [table= " + em.fullTableName() + ", key=" + key + "]");
-
- Connection conn = null;
-
- PreparedStatement stmt = null;
-
- try {
- conn = connection();
-
- stmt = conn.prepareStatement(em.loadQrySingle);
-
- fillKeyParameters(stmt, em, key);
-
- ResultSet rs = stmt.executeQuery();
-
- if (rs.next())
- return buildObject(em.cacheName, em.valueType(), em.valueColumns(), em.loadColIdxs, rs);
- }
- catch (SQLException e) {
- throw new CacheLoaderException("Failed to load object [table=" + em.fullTableName() +
- ", key=" + key + "]", e);
- }
- finally {
- end(conn, stmt);
- }
-
- return null;
- }
-
- /** {@inheritDoc} */
- @Override public Map<K, V> loadAll(Iterable<? extends K> keys) throws CacheLoaderException {
- assert keys != null;
-
- Connection conn = null;
-
- try {
- conn = connection();
-
- String cacheName = session().cacheName();
-
- Map<Object, LoadWorker<K, V>> workers = U.newHashMap(cacheMappings(cacheName).size());
-
- Map<K, V> res = new HashMap<>();
-
- for (K key : keys) {
- Object keyTypeId = typeIdForObject(key);
-
- EntryMapping em = entryMapping(cacheName, keyTypeId, key);
-
- LoadWorker<K, V> worker = workers.get(keyTypeId);
-
- if (worker == null)
- workers.put(keyTypeId, worker = new LoadWorker<>(conn, em));
-
- worker.keys.add(key);
-
- if (worker.keys.size() == em.maxKeysPerStmt)
- res.putAll(workers.remove(keyTypeId).call());
- }
-
- for (LoadWorker<K, V> worker : workers.values())
- res.putAll(worker.call());
-
- return res;
- }
- catch (Exception e) {
- throw new CacheWriterException("Failed to load entries from database", e);
- }
- finally {
- closeConnection(conn);
- }
- }
-
- /**
- * @param insStmt Insert statement.
- * @param updStmt Update statement.
- * @param em Entry mapping.
- * @param entry Cache entry.
- * @throws CacheWriterException If failed to update record in database.
- */
- private void writeUpsert(PreparedStatement insStmt, PreparedStatement updStmt,
- EntryMapping em, Cache.Entry<? extends K, ? extends V> entry) throws CacheWriterException {
- try {
- CacheWriterException we = null;
-
- for (int attempt = 0; attempt < maxWrtAttempts; attempt++) {
- int paramIdx = fillValueParameters(updStmt, 1, em, entry.getValue());
-
- fillKeyParameters(updStmt, paramIdx, em, entry.getKey());
-
- if (updStmt.executeUpdate() == 0) {
- paramIdx = fillKeyParameters(insStmt, em, entry.getKey());
-
- fillValueParameters(insStmt, paramIdx, em, entry.getValue());
-
- try {
- insStmt.executeUpdate();
-
- if (attempt > 0)
- U.warn(log, "Entry was inserted in database on second try [table=" + em.fullTableName() +
- ", entry=" + entry + "]");
- }
- catch (SQLException e) {
- String sqlState = e.getSQLState();
-
- SQLException nested = e.getNextException();
-
- while (sqlState == null && nested != null) {
- sqlState = nested.getSQLState();
-
- nested = nested.getNextException();
- }
-
- // The error with code 23505 or 23000 is thrown when trying to insert a row that
- // would violate a unique index or primary key.
- if ("23505".equals(sqlState) || "23000".equals(sqlState)) {
- if (we == null)
- we = new CacheWriterException("Failed insert entry in database, violate a unique" +
- " index or primary key [table=" + em.fullTableName() + ", entry=" + entry + "]");
-
- we.addSuppressed(e);
-
- U.warn(log, "Failed insert entry in database, violate a unique index or primary key" +
- " [table=" + em.fullTableName() + ", entry=" + entry + "]");
-
- continue;
- }
-
- throw new CacheWriterException("Failed insert entry in database [table=" + em.fullTableName() +
- ", entry=" + entry, e);
- }
- }
-
- if (attempt > 0)
- U.warn(log, "Entry was updated in database on second try [table=" + em.fullTableName() +
- ", entry=" + entry + "]");
-
- return;
- }
-
- throw we;
- }
- catch (SQLException e) {
- throw new CacheWriterException("Failed update entry in database [table=" + em.fullTableName() +
- ", entry=" + entry + "]", e);
- }
- }
-
- /** {@inheritDoc} */
- @Override public void write(Cache.Entry<? extends K, ? extends V> entry) throws CacheWriterException {
- assert entry != null;
-
- K key = entry.getKey();
-
- EntryMapping em = entryMapping(session().cacheName(), typeIdForObject(key), key);
-
- if (log.isDebugEnabled())
- log.debug("Start write entry to database [table=" + em.fullTableName() + ", entry=" + entry + "]");
-
- Connection conn = null;
-
- try {
- conn = connection();
-
- if (dialect.hasMerge()) {
- PreparedStatement stmt = null;
-
- try {
- stmt = conn.prepareStatement(em.mergeQry);
-
- int i = fillKeyParameters(stmt, em, key);
-
- fillValueParameters(stmt, i, em, entry.getValue());
-
- int updCnt = stmt.executeUpdate();
-
- if (updCnt != 1)
- U.warn(log, "Unexpected number of updated entries [table=" + em.fullTableName() +
- ", entry=" + entry + "expected=1, actual=" + updCnt + "]");
- }
- finally {
- U.closeQuiet(stmt);
- }
- }
- else {
- PreparedStatement insStmt = null;
-
- PreparedStatement updStmt = null;
-
- try {
- insStmt = conn.prepareStatement(em.insQry);
-
- updStmt = conn.prepareStatement(em.updQry);
-
- writeUpsert(insStmt, updStmt, em, entry);
- }
- finally {
- U.closeQuiet(insStmt);
-
- U.closeQuiet(updStmt);
- }
- }
- }
- catch (SQLException e) {
- throw new CacheWriterException("Failed to write entry to database [table=" + em.fullTableName() +
- ", entry=" + entry + "]", e);
- }
- finally {
- closeConnection(conn);
- }
- }
-
- /** {@inheritDoc} */
- @Override public void writeAll(final Collection<Cache.Entry<? extends K, ? extends V>> entries)
- throws CacheWriterException {
- assert entries != null;
-
- Connection conn = null;
-
- try {
- conn = connection();
-
- String cacheName = session().cacheName();
-
- Object currKeyTypeId = null;
-
- if (dialect.hasMerge()) {
- PreparedStatement mergeStmt = null;
-
- try {
- EntryMapping em = null;
-
- LazyValue<Object[]> lazyEntries = new LazyValue<Object[]>() {
- @Override public Object[] create() {
- return entries.toArray();
- }
- };
-
- int fromIdx = 0, prepared = 0;
-
- for (Cache.Entry<? extends K, ? extends V> entry : entries) {
- K key = entry.getKey();
-
- Object keyTypeId = typeIdForObject(key);
-
- em = entryMapping(cacheName, keyTypeId, key);
-
- if (currKeyTypeId == null || !currKeyTypeId.equals(keyTypeId)) {
- if (mergeStmt != null) {
- if (log.isDebugEnabled())
- log.debug("Write entries to db [cache name=" + cacheName +
- ", key type=" + em.keyType() + ", count=" + prepared + "]");
-
- executeBatch(em, mergeStmt, "writeAll", fromIdx, prepared, lazyEntries);
-
- U.closeQuiet(mergeStmt);
- }
-
- mergeStmt = conn.prepareStatement(em.mergeQry);
-
- currKeyTypeId = keyTypeId;
-
- fromIdx += prepared;
-
- prepared = 0;
- }
-
- int i = fillKeyParameters(mergeStmt, em, key);
-
- fillValueParameters(mergeStmt, i, em, entry.getValue());
-
- mergeStmt.addBatch();
-
- if (++prepared % batchSz == 0) {
- if (log.isDebugEnabled())
- log.debug("Write entries to db [cache name=" + cacheName +
- ", key type=" + em.keyType() + ", count=" + prepared + "]");
-
- executeBatch(em, mergeStmt, "writeAll", fromIdx, prepared, lazyEntries);
-
- fromIdx += prepared;
-
- prepared = 0;
- }
- }
-
- if (mergeStmt != null && prepared % batchSz != 0) {
- if (log.isDebugEnabled())
- log.debug("Write entries to db [cache name=" + cacheName +
- ", key type=" + em.keyType() + ", count=" + prepared + "]");
-
- executeBatch(em, mergeStmt, "writeAll", fromIdx, prepared, lazyEntries);
-
- }
- }
- finally {
- U.closeQuiet(mergeStmt);
- }
- }
- else {
- if (log.isDebugEnabled())
- log.debug("Write entries to db one by one using update and insert statements [cache name=" +
- cacheName + ", count=" + entries.size() + "]");
-
- PreparedStatement insStmt = null;
-
- PreparedStatement updStmt = null;
-
- try {
- for (Cache.Entry<? extends K, ? extends V> entry : entries) {
- K key = entry.getKey();
-
- Object keyTypeId = typeIdForObject(key);
-
- EntryMapping em = entryMapping(cacheName, keyTypeId, key);
-
- 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);
- }
- }
- }
- catch (SQLException e) {
- throw new CacheWriterException("Failed to write entries in database", e);
- }
- finally {
- closeConnection(conn);
- }
- }
-
- /** {@inheritDoc} */
- @Override public void delete(Object key) throws CacheWriterException {
- assert key != null;
-
- EntryMapping em = entryMapping(session().cacheName(), typeIdForObject(key), key);
-
- if (log.isDebugEnabled())
- log.debug("Remove value from db [table=" + em.fullTableName() + ", key=" + key + "]");
-
- Connection conn = null;
-
- PreparedStatement stmt = null;
-
- try {
- conn = connection();
-
- stmt = conn.prepareStatement(em.remQry);
-
- fillKeyParameters(stmt, em, key);
-
- int delCnt = stmt.executeUpdate();
-
- if (delCnt != 1)
- U.warn(log, "Unexpected number of deleted entries [table=" + em.fullTableName() + ", key=" + key +
- ", expected=1, actual=" + delCnt + "]");
- }
- catch (SQLException e) {
- throw new CacheWriterException("Failed to remove value from database [table=" + em.fullTableName() +
- ", key=" + key + "]", e);
- }
- finally {
- end(conn, stmt);
- }
- }
-
- /**
- * @param em Entry mapping.
- * @param stmt Statement.
- * @param desc Statement description for error message.
- * @param fromIdx Objects in batch start from index.
- * @param prepared Expected objects in batch.
- * @param lazyObjs All objects used in batch statement as array.
- * @throws SQLException If failed to execute batch statement.
- */
- private void executeBatch(EntryMapping em, Statement stmt, String desc, int fromIdx, int prepared,
- LazyValue<Object[]> lazyObjs) throws SQLException {
- try {
- int[] rowCounts = stmt.executeBatch();
-
- int numOfRowCnt = rowCounts.length;
-
- if (numOfRowCnt != prepared)
- U.warn(log, "Unexpected number of updated rows [table=" + em.fullTableName() + ", expected=" + prepared +
- ", actual=" + numOfRowCnt + "]");
-
- for (int i = 0; i < numOfRowCnt; i++) {
- int cnt = rowCounts[i];
-
- if (cnt != 1 && cnt != SUCCESS_NO_INFO) {
- Object[] objs = lazyObjs.value();
-
- U.warn(log, "Batch " + desc + " returned unexpected updated row count [table=" + em.fullTableName() +
- ", entry=" + objs[fromIdx + i] + ", expected=1, actual=" + cnt + "]");
- }
- }
- }
- catch (BatchUpdateException be) {
- int[] rowCounts = be.getUpdateCounts();
-
- for (int i = 0; i < rowCounts.length; i++) {
- if (rowCounts[i] == EXECUTE_FAILED) {
- Object[] objs = lazyObjs.value();
-
- U.warn(log, "Batch " + desc + " failed on execution [table=" + em.fullTableName() +
- ", entry=" + objs[fromIdx + i] + "]");
- }
- }
-
- throw be;
- }
- }
-
- /** {@inheritDoc} */
- @Override public void deleteAll(final Collection<?> keys) throws CacheWriterException {
- assert keys != null;
-
- Connection conn = null;
-
- try {
- conn = connection();
-
- LazyValue<Object[]> lazyKeys = new LazyValue<Object[]>() {
- @Override public Object[] create() {
- return keys.toArray();
- }
- };
-
- String cacheName = session().cacheName();
-
- Object currKeyTypeId = null;
-
- EntryMapping em = null;
-
- PreparedStatement delStmt = null;
-
- int fromIdx = 0, prepared = 0;
-
- for (Object key : keys) {
- Object keyTypeId = typeIdForObject(key);
-
- em = entryMapping(cacheName, keyTypeId, key);
-
- if (delStmt == null) {
- delStmt = conn.prepareStatement(em.remQry);
-
- currKeyTypeId = keyTypeId;
- }
-
- if (!currKeyTypeId.equals(keyTypeId)) {
- if (log.isDebugEnabled())
- log.debug("Delete entries from db [cache name=" + cacheName +
- ", key type=" + em.keyType() + ", count=" + prepared + "]");
-
- executeBatch(em, delStmt, "deleteAll", fromIdx, prepared, lazyKeys);
-
- fromIdx += prepared;
-
- prepared = 0;
-
- currKeyTypeId = keyTypeId;
- }
-
- fillKeyParameters(delStmt, em, key);
-
- delStmt.addBatch();
-
- if (++prepared % batchSz == 0) {
- if (log.isDebugEnabled())
- log.debug("Delete entries from db [cache name=" + cacheName +
- ", key type=" + em.keyType() + ", count=" + prepared + "]");
-
- executeBatch(em, delStmt, "deleteAll", fromIdx, prepared, lazyKeys);
-
- fromIdx += prepared;
-
- prepared = 0;
- }
- }
-
- if (delStmt != null && prepared % batchSz != 0) {
- if (log.isDebugEnabled())
- log.debug("Delete entries from db [cache name=" + cacheName +
- ", key type=" + em.keyType() + ", count=" + prepared + "]");
-
- executeBatch(em, delStmt, "deleteAll", fromIdx, prepared, lazyKeys);
- }
- }
- catch (SQLException e) {
- throw new CacheWriterException("Failed to remove values from database", e);
- }
- finally {
- closeConnection(conn);
- }
- }
-
- /**
- * Sets the value of the designated parameter using the given object.
- *
- * @param stmt Prepare statement.
- * @param i Index for parameters.
- * @param field Field descriptor.
- * @param fieldVal Field value.
- * @throws CacheException If failed to set statement parameter.
- */
- private void fillParameter(PreparedStatement stmt, int i, CacheJdbcPojoStoreTypeField field, @Nullable Object fieldVal)
- throws CacheException {
- try {
- if (fieldVal != null) {
- if (field.getJavaFieldType() == UUID.class) {
- switch (field.getDatabaseFieldType()) {
- case Types.BINARY:
- fieldVal = U.uuidToBytes((UUID)fieldVal);
-
- break;
- case Types.CHAR:
- case Types.VARCHAR:
- fieldVal = fieldVal.toString();
-
- break;
- }
- }
-
- stmt.setObject(i, fieldVal);
- }
- else
- stmt.setNull(i, field.getDatabaseFieldType());
- }
- catch (SQLException e) {
- throw new CacheException("Failed to set statement parameter name: " + field.getDatabaseFieldName(), e);
- }
- }
-
- /**
- * @param stmt Prepare statement.
- * @param idx Start index for parameters.
- * @param em Entry mapping.
- * @param key Key object.
- * @return Next index for parameters.
- * @throws CacheException If failed to set statement parameters.
- */
- 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.getJavaFieldName(), key);
-
- fillParameter(stmt, idx++, field, fieldVal);
- }
-
- return idx;
- }
-
- /**
- * @param stmt Prepare statement.
- * @param m Type mapping description.
- * @param key Key object.
- * @return Next index for parameters.
- * @throws CacheException If failed to set statement parameters.
- */
- private int fillKeyParameters(PreparedStatement stmt, EntryMapping m, Object key) throws CacheException {
- return fillKeyParameters(stmt, 1, m, key);
- }
-
- /**
- * @param stmt Prepare statement.
- * @param idx Start index for parameters.
- * @param em Type mapping description.
- * @param val Value object.
- * @return Next index for parameters.
- * @throws CacheException If failed to set statement parameters.
- */
- 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.getJavaFieldName(), val);
-
- fillParameter(stmt, idx++, field, fieldVal);
- }
-
- return idx;
- }
-
- /**
- * @return Data source.
- */
- public DataSource getDataSource() {
- return dataSrc;
- }
-
- /**
- * @param dataSrc Data source.
- */
- public void setDataSource(DataSource dataSrc) {
- this.dataSrc = dataSrc;
- }
-
- /**
- * Get database dialect.
- *
- * @return Database dialect.
- */
- public JdbcDialect getDialect() {
- return dialect;
- }
-
- /**
- * Set database dialect.
- *
- * @param dialect Database dialect.
- */
- public void setDialect(JdbcDialect dialect) {
- this.dialect = dialect;
- }
-
- /**
- * Get Max workers thread count. These threads are responsible for execute query.
- *
- * @return Max workers thread count.
- */
- public int getMaximumPoolSize() {
- return maxPoolSz;
- }
-
- /**
- * Gets maximum number of write attempts in case of database error.
- *
- * @return Maximum number of write attempts.
- */
- public int getMaximumWriteAttempts() {
- return maxWrtAttempts;
- }
-
- /**
- * Sets maximum number of write attempts in case of database error.
- *
- * @param maxWrtAttempts Number of write attempts.
- * @return {@code This} for chaining.
- */
- public CacheJdbcPojoStore<K, V> setMaximumWriteAttempts(int maxWrtAttempts) {
- this.maxWrtAttempts = maxWrtAttempts;
-
- return this;
- }
-
- /**
- * Gets types known by store.
- *
- * @return Types known by store.
- */
- public CacheJdbcPojoStoreType[] getTypes() {
- return types;
- }
-
- /**
- * Sets store configurations.
- *
- * @param types Store should process.
- * @return {@code This} for chaining.
- */
- public CacheJdbcPojoStore<K, V> setTypes(CacheJdbcPojoStoreType... types) {
- this.types = types;
-
- return this;
- }
-
- /**
- * Set Max workers thread count. These threads are responsible for execute query.
- *
- * @param maxPoolSz Max workers thread count.
- */
- public void setMaximumPoolSize(int maxPoolSz) {
- this.maxPoolSz = maxPoolSz;
- }
-
- /**
- * Get maximum batch size for delete and delete operations.
- *
- * @return Maximum batch size.
- */
- public int getBatchSize() {
- return batchSz;
- }
-
- /**
- * Set maximum batch size for write and delete operations.
- *
- * @param batchSz Maximum batch size.
- */
- public void setBatchSize(int batchSz) {
- this.batchSz = batchSz;
- }
-
- /**
- * Parallel load cache minimum row count threshold.
- *
- * @return If {@code 0} then load sequentially.
- */
- public int getParallelLoadCacheMinimumThreshold() {
- return parallelLoadCacheMinThreshold;
- }
-
- /**
- * Parallel load cache minimum row count threshold.
- *
- * @param parallelLoadCacheMinThreshold Minimum row count threshold. If {@code 0} then load sequentially.
- */
- public void setParallelLoadCacheMinimumThreshold(int parallelLoadCacheMinThreshold) {
- this.parallelLoadCacheMinThreshold = parallelLoadCacheMinThreshold;
- }
-
- /**
- * @return Ignite instance.
- */
- private Ignite ignite() {
- return ignite;
- }
-
- /**
- * @return Store session.
- */
- private CacheStoreSession session() {
- return ses;
- }
-
- /**
- * POJO methods cache.
- */
- private static class PojoMethodsCache {
- /** POJO class. */
- private final Class<?> cls;
-
- /** Constructor for POJO object. */
- private Constructor ctor;
-
- /** {@code true} if object is a simple type. */
- private final boolean simple;
-
- /** Cached setters for POJO object. */
- private Map<String, Method> getters;
-
- /** Cached getters for POJO object. */
- private Map<String, Method> setters;
-
- /**
- * POJO methods cache.
- *
- * @param clsName Class name.
- * @param fields Fields.
- * @throws CacheException If failed to construct type cache.
- */
- public PojoMethodsCache(String clsName, CacheJdbcPojoStoreTypeField[] fields) throws CacheException {
- try {
- cls = Class.forName(clsName);
-
- if (simple = simpleType(cls))
- return;
-
- ctor = cls.getDeclaredConstructor();
-
- if (!ctor.isAccessible())
- ctor.setAccessible(true);
- }
- catch (ClassNotFoundException e) {
- throw new CacheException("Failed to find class: " + clsName, e);
- }
- catch (NoSuchMethodException e) {
- throw new CacheException("Failed to find default constructor for class: " + clsName, e);
- }
-
- setters = U.newHashMap(fields.length);
-
- getters = U.newHashMap(fields.length);
-
- for (CacheJdbcPojoStoreTypeField field : fields) {
- String prop = capitalFirst(field.getJavaFieldName());
-
- try {
- getters.put(field.getJavaFieldName(), cls.getMethod("get" + prop));
- }
- catch (NoSuchMethodException ignored) {
- try {
- getters.put(field.getJavaFieldName(), cls.getMethod("is" + prop));
- }
- catch (NoSuchMethodException e) {
- throw new CacheException("Failed to find getter in POJO class [clsName=" + clsName +
- ", prop=" + field.getJavaFieldName() + "]", e);
- }
- }
-
- try {
- setters.put(field.getJavaFieldName(), cls.getMethod("set" + prop, field.getJavaFieldType()));
- }
- catch (NoSuchMethodException e) {
- throw new CacheException("Failed to find setter in POJO class [clsName=" + clsName +
- ", prop=" + field.getJavaFieldName() + "]", e);
+ catch (NoSuchMethodException e) {
+ throw new CacheException("Failed to find setter in POJO class [clsName=" + clsName +
+ ", prop=" + field.getJavaFieldName() + "]", e);
}
}
}
@@ -1925,367 +465,4 @@ public class CacheJdbcPojoStore<K, V> implements CacheStore<K, V>, LifecycleAwar
str.isEmpty() ? "" : Character.toUpperCase(str.charAt(0)) + str.substring(1);
}
}
-
- /**
- * Entry mapping description.
- */
- private static class EntryMapping {
- /** Cache name. */
- private final String cacheName;
-
- /** Database dialect. */
- private final JdbcDialect dialect;
-
- /** Select border for range queries. */
- private final String loadCacheSelRangeQry;
-
- /** Select all items query. */
- private final String loadCacheQry;
-
- /** Select item query. */
- private final String loadQrySingle;
-
- /** Select items query. */
- private final String loadQry;
-
- /** Merge item(s) query. */
- private final String mergeQry;
-
- /** Update item query. */
- private final String insQry;
-
- /** Update item query. */
- private final String updQry;
-
- /** Remove item(s) query. */
- private final String remQry;
-
- /** Max key count for load query per statement. */
- private final int maxKeysPerStmt;
-
- /** Database key columns. */
- private final Collection<String> keyCols;
-
- /** Database unique value columns. */
- private final Collection<String> cols;
-
- /** Select query columns index. */
- private final Map<String, Integer> loadColIdxs;
-
- /** Unique value fields. */
- private final Collection<CacheJdbcPojoStoreTypeField> uniqValFields;
-
- /** Type metadata. */
- private final CacheJdbcPojoStoreType typeMeta;
-
- /** Full table name. */
- private final String fullTblName;
-
- /**
- * @param cacheName Cache name.
- * @param dialect JDBC dialect.
- * @param typeMeta Type metadata.
- */
- public EntryMapping(@Nullable String cacheName, JdbcDialect dialect, CacheJdbcPojoStoreType typeMeta) {
- this.cacheName = cacheName;
-
- this.dialect = dialect;
-
- this.typeMeta = typeMeta;
-
- CacheJdbcPojoStoreTypeField[] keyFields = typeMeta.getKeyFields();
-
- CacheJdbcPojoStoreTypeField[] valFields = typeMeta.getValueFields();
-
- keyCols = databaseColumns(F.asList(keyFields));
-
- uniqValFields = F.view(F.asList(valFields), new IgnitePredicate<CacheJdbcPojoStoreTypeField>() {
- @Override public boolean apply(CacheJdbcPojoStoreTypeField col) {
- return !keyCols.contains(col.getDatabaseFieldName());
- }
- });
-
- String schema = typeMeta.getDatabaseSchema();
-
- String tblName = typeMeta.getDatabaseTable();
-
- fullTblName = F.isEmpty(schema) ? tblName : schema + "." + tblName;
-
- Collection<String> uniqValCols = databaseColumns(uniqValFields);
-
- cols = F.concat(false, keyCols, uniqValCols);
-
- loadColIdxs = U.newHashMap(cols.size());
-
- int idx = 1;
-
- for (String col : cols)
- loadColIdxs.put(col, idx++);
-
- loadCacheQry = dialect.loadCacheQuery(fullTblName, cols);
-
- loadCacheSelRangeQry = dialect.loadCacheSelectRangeQuery(fullTblName, keyCols);
-
- loadQrySingle = dialect.loadQuery(fullTblName, keyCols, cols, 1);
-
- maxKeysPerStmt = dialect.getMaxParameterCount() / keyCols.size();
-
- loadQry = dialect.loadQuery(fullTblName, keyCols, cols, maxKeysPerStmt);
-
- insQry = dialect.insertQuery(fullTblName, keyCols, uniqValCols);
-
- updQry = dialect.updateQuery(fullTblName, keyCols, uniqValCols);
-
- mergeQry = dialect.mergeQuery(fullTblName, keyCols, uniqValCols);
-
- remQry = dialect.removeQuery(fullTblName, keyCols);
- }
-
- /**
- * Extract database column names from {@link CacheJdbcPojoStoreTypeField}.
- *
- * @param dsc collection of {@link CacheJdbcPojoStoreTypeField}.
- * @return Collection with database column names.
- */
- private static Collection<String> databaseColumns(Collection<CacheJdbcPojoStoreTypeField> dsc) {
- return F.transform(dsc, new C1<CacheJdbcPojoStoreTypeField, String>() {
- /** {@inheritDoc} */
- @Override public String apply(CacheJdbcPojoStoreTypeField col) {
- return col.getDatabaseFieldName();
- }
- });
- }
-
- /**
- * Construct query for select values with key count less or equal {@code maxKeysPerStmt}
- *
- * @param keyCnt Key count.
- * @return Load query statement text.
- */
- private String loadQuery(int keyCnt) {
- assert keyCnt <= maxKeysPerStmt;
-
- if (keyCnt == maxKeysPerStmt)
- return loadQry;
-
- if (keyCnt == 1)
- return loadQrySingle;
-
- return dialect.loadQuery(fullTblName, keyCols, cols, keyCnt);
- }
-
- /**
- * Construct query for select values in range.
- *
- * @param appendLowerBound Need add lower bound for range.
- * @param appendUpperBound Need add upper bound for range.
- * @return Query with range.
- */
- private String loadCacheRangeQuery(boolean appendLowerBound, boolean appendUpperBound) {
- return dialect.loadCacheRangeQuery(fullTblName, keyCols, cols, appendLowerBound, appendUpperBound);
- }
-
- /**
- * @return Key type.
- */
- protected String keyType() {
- return typeMeta.getKeyType();
- }
-
- /**
- * @return Value type.
- */
- protected String valueType() {
- return typeMeta.getValueType();
- }
-
- /**
- * Gets key columns.
- *
- * @return Key columns.
- */
- protected CacheJdbcPojoStoreTypeField[] keyColumns() {
- return typeMeta.getKeyFields();
- }
-
- /**
- * Gets value columns.
- *
- * @return Value columns.
- */
- protected CacheJdbcPojoStoreTypeField[] valueColumns() {
- return typeMeta.getValueFields();
- }
-
- /**
- * Get full table name.
- *
- * @return <schema>.<table name>
- */
- protected String fullTableName() {
- return fullTblName;
- }
- }
-
- /**
- * Worker for load cache using custom user query.
- *
- * @param <K1> Key type.
- * @param <V1> Value type.
- */
- private class LoadCacheCustomQueryWorker<K1, V1> implements Callable<Void> {
- /** Entry mapping description. */
- private final EntryMapping em;
-
- /** User query. */
- private final String qry;
-
- /** Closure for loaded values. */
- private final IgniteBiInClosure<K1, V1> clo;
-
- /**
- * @param em Entry mapping description.
- * @param qry User query.
- * @param clo Closure for loaded values.
- */
- private LoadCacheCustomQueryWorker(EntryMapping em, String qry, IgniteBiInClosure<K1, V1> clo) {
- this.em = em;
- this.qry = qry;
- this.clo = clo;
- }
-
- /** {@inheritDoc} */
- @Override public Void call() throws Exception {
- if (log.isDebugEnabled())
- log.debug("Load cache using custom query [cache name= " + em.cacheName +
- ", key type=" + em.keyType() + ", query=" + qry + "]");
-
- Connection conn = null;
-
- PreparedStatement stmt = null;
-
- try {
- conn = openConnection(true);
-
- stmt = conn.prepareStatement(qry);
-
- ResultSet rs = stmt.executeQuery();
-
- ResultSetMetaData meta = rs.getMetaData();
-
- Map<String, Integer> colIdxs = U.newHashMap(meta.getColumnCount());
-
- for (int i = 1; i <= meta.getColumnCount(); i++)
- colIdxs.put(meta.getColumnLabel(i), i);
-
- while (rs.next()) {
- K1 key = buildObject(em.cacheName, em.keyType(), em.keyColumns(), colIdxs, rs);
- V1 val = buildObject(em.cacheName, em.valueType(), em.valueColumns(), colIdxs, rs);
-
- clo.apply(key, val);
- }
-
- return null;
- }
- catch (SQLException e) {
- throw new CacheLoaderException("Failed to execute custom query for load cache", e);
- }
- finally {
- U.closeQuiet(stmt);
-
- U.closeQuiet(conn);
- }
- }
- }
-
- /**
- * Lazy initialization of value.
- *
- * @param <T> Cached object type
- */
- private abstract static class LazyValue<T> {
- /** Cached value. */
- private T val;
-
- /**
- * @return Construct value.
- */
- protected abstract T create();
-
- /**
- * @return Value.
- */
- public T value() {
- if (val == null)
- val = create();
-
- return val;
- }
- }
-
- /**
- * Worker for load by keys.
- *
- * @param <K1> Key type.
- * @param <V1> Value type.
- */
- private class LoadWorker<K1, V1> implements Callable<Map<K1, V1>> {
- /** Connection. */
- private final Connection conn;
-
- /** Keys for load. */
- private final Collection<K1> keys;
-
- /** Entry mapping description. */
- private final EntryMapping em;
-
- /**
- * @param conn Connection.
- * @param em Entry mapping description.
- */
- private LoadWorker(Connection conn, EntryMapping em) {
- this.conn = conn;
- this.em = em;
-
- keys = new ArrayList<>(em.maxKeysPerStmt);
- }
-
- /** {@inheritDoc} */
- @Override public Map<K1, V1> call() throws Exception {
- if (log.isDebugEnabled())
- log.debug("Load values from db [table= " + em.fullTableName() +
- ", key count=" + keys.size() + "]");
-
- PreparedStatement stmt = null;
-
- try {
- stmt = conn.prepareStatement(em.loadQuery(keys.size()));
-
- int idx = 1;
-
- for (Object key : keys)
- for (CacheJdbcPojoStoreTypeField field : em.keyColumns()) {
- Object fieldVal = extractParameter(em.cacheName, em.keyType(), field.getJavaFieldName(), key);
-
- fillParameter(stmt, idx++, field, fieldVal);
- }
-
- ResultSet rs = stmt.executeQuery();
-
- Map<K1, V1> entries = U.newHashMap(keys.size());
-
- while (rs.next()) {
- K1 key = buildObject(em.cacheName, em.keyType(), em.keyColumns(), em.loadColIdxs, rs);
- V1 val = buildObject(em.cacheName, em.valueType(), em.valueColumns(), em.loadColIdxs, rs);
-
- entries.put(key, val);
- }
-
- return entries;
- }
- finally {
- U.closeQuiet(stmt);
- }
- }
- }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2961aaa7/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 b333bc7..206f1fe 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
@@ -56,7 +56,7 @@ public class CacheJdbcPojoStoreConfiguration implements Serializable {
private int parallelLoadCacheMinThreshold = DFLT_PARALLEL_LOAD_CACHE_MINIMUM_THRESHOLD;
/** Types that store could process. */
- private CacheJdbcPojoStoreType[] types;
+ private JdbcType[] types;
/**
* Empty constructor (all values are initialized to their defaults).
@@ -212,7 +212,7 @@ public class CacheJdbcPojoStoreConfiguration implements Serializable {
*
* @return Types known by store.
*/
- public CacheJdbcPojoStoreType[] getTypes() {
+ public JdbcType[] getTypes() {
return types;
}
@@ -222,7 +222,7 @@ public class CacheJdbcPojoStoreConfiguration implements Serializable {
* @param types Store should process.
* @return {@code This} for chaining.
*/
- public CacheJdbcPojoStoreConfiguration setTypes(CacheJdbcPojoStoreType... types) {
+ public CacheJdbcPojoStoreConfiguration setTypes(JdbcType... types) {
this.types = types;
return this;
http://git-wip-us.apache.org/repos/asf/ignite/blob/2961aaa7/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
deleted file mode 100644
index e755165..0000000
--- a/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheJdbcPojoStoreType.java
+++ /dev/null
@@ -1,272 +0,0 @@
-/*
- * 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 java.io.Serializable;
-import org.apache.ignite.internal.util.tostring.GridToStringInclude;
-
-/**
- * Description for type that could be stored into database by store.
- */
-public class CacheJdbcPojoStoreType implements Serializable {
- /** */
- private static final long serialVersionUID = 0L;
-
- /** Cache name. */
- private String cacheName;
-
- /** Schema name in database. */
- private String dbSchema;
-
- /** Table name in database. */
- private String dbTbl;
-
- /** Key class used to store key in cache. */
- private String keyType;
-
- /** List of fields descriptors for key object. */
- @GridToStringInclude
- private CacheJdbcPojoStoreTypeField[] keyFields;
-
- /** Value class used to store value in cache. */
- private String valType;
-
- /** List of fields descriptors for value object. */
- @GridToStringInclude
- private CacheJdbcPojoStoreTypeField[] valFields;
-
- /** If {@code true} object is stored as IgniteObject. */
- private boolean keepSerialized;
-
- /**
- * Empty constructor (all values are initialized to their defaults).
- */
- public CacheJdbcPojoStoreType() {
- /* No-op. */
- }
-
- /**
- * Copy constructor.
- *
- * @param type Type to copy.
- */
- public CacheJdbcPojoStoreType(CacheJdbcPojoStoreType type) {
- cacheName = type.getCacheName();
-
- dbSchema = type.getDatabaseSchema();
- dbTbl = type.getDatabaseTable();
-
- keyType = type.getKeyType();
- keyFields = type.getKeyFields();
-
- valType = type.getValueType();
- valFields = type.getValueFields();
-
- keepSerialized = type.isKeepSerialized();
- }
-
- /**
- * Gets associated cache name.
- *
- * @return Сache name.
- */
- public String getCacheName() {
- return cacheName;
- }
-
- /**
- * Sets associated cache name.
- *
- * @param cacheName Cache name.
- */
- public CacheJdbcPojoStoreType setCacheName(String cacheName) {
- this.cacheName = cacheName;
-
- return this;
- }
-
- /**
- * Gets database schema name.
- *
- * @return Schema name.
- */
- public String getDatabaseSchema() {
- return dbSchema;
- }
-
- /**
- * Sets database schema name.
- *
- * @param dbSchema Schema name.
- */
- public CacheJdbcPojoStoreType setDatabaseSchema(String dbSchema) {
- this.dbSchema = dbSchema;
-
- return this;
- }
-
- /**
- * Gets table name in database.
- *
- * @return Table name in database.
- */
- public String getDatabaseTable() {
- return dbTbl;
- }
-
- /**
- * Table name in database.
- *
- * @param dbTbl Table name in database.
- * @return {@code this} for chaining.
- */
- public CacheJdbcPojoStoreType setDatabaseTable(String dbTbl) {
- this.dbTbl = dbTbl;
-
- return this;
- }
-
- /**
- * Gets key type.
- *
- * @return Key type.
- */
- public String getKeyType() {
- return keyType;
- }
-
- /**
- * Sets key type.
- *
- * @param keyType Key type.
- * @return {@code this} for chaining.
- */
- public CacheJdbcPojoStoreType setKeyType(String keyType) {
- this.keyType = keyType;
-
- return this;
- }
-
- /**
- * Sets key type.
- *
- * @param cls Key type class.
- * @return {@code this} for chaining.
- */
- public CacheJdbcPojoStoreType setKeyType(Class<?> cls) {
- setKeyType(cls.getName());
-
- return this;
- }
-
- /**
- * Gets value type.
- *
- * @return Key type.
- */
- public String getValueType() {
- return valType;
- }
-
- /**
- * Sets value type.
- *
- * @param valType Value type.
- * @return {@code this} for chaining.
- */
- public CacheJdbcPojoStoreType setValueType(String valType) {
- this.valType = valType;
-
- return this;
- }
-
- /**
- * Sets value type.
- *
- * @param cls Value type class.
- * @return {@code this} for chaining.
- */
- public CacheJdbcPojoStoreType setValueType(Class<?> cls) {
- setValueType(cls.getName());
-
- return this;
- }
-
- /**
- * Gets optional persistent key fields (needed only if {@link CacheJdbcPojoStore} is used).
- *
- * @return Persistent key fields.
- */
- public CacheJdbcPojoStoreTypeField[] getKeyFields() {
- return keyFields;
- }
-
- /**
- * Sets optional persistent key fields (needed only if {@link CacheJdbcPojoStore} is used).
- *
- * @param keyFields Persistent key fields.
- * @return {@code this} for chaining.
- */
- public CacheJdbcPojoStoreType setKeyFields(CacheJdbcPojoStoreTypeField... keyFields) {
- this.keyFields = keyFields;
-
- return this;
- }
-
- /**
- * Gets optional persistent value fields (needed only if {@link CacheJdbcPojoStore} is used).
- *
- * @return Persistent value fields.
- */
- public CacheJdbcPojoStoreTypeField[] getValueFields() {
- return valFields;
- }
-
- /**
- * Sets optional persistent value fields (needed only if {@link CacheJdbcPojoStore} is used).
- *
- * @param valFields Persistent value fields.
- * @return {@code this} for chaining.
- */
- public CacheJdbcPojoStoreType setValueFields(CacheJdbcPojoStoreTypeField... valFields) {
- this.valFields = valFields;
-
- return this;
- }
-
- /**
- * Gets how value stored in cache.
- *
- * @return {@code true} if object is stored as IgniteObject.
- */
- public boolean isKeepSerialized() {
- return keepSerialized;
- }
-
- /**
- * Sets how value stored in cache.
- *
- * @param keepSerialized {@code true} if object is stored as IgniteObject.
- * @return {@code this} for chaining.
- */
- public CacheJdbcPojoStoreType setKeepSerialized(boolean keepSerialized) {
- this.keepSerialized = keepSerialized;
-
- return this;
- }
-}
[3/4] ignite git commit: IGNITE-1753 Wip reworking code.
Posted by ak...@apache.org.
IGNITE-1753 Wip reworking code.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/2961aaa7
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/2961aaa7
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/2961aaa7
Branch: refs/heads/ignite-1753-1282
Commit: 2961aaa72ad587fd1578983551c4397a8fdb3b3d
Parents: 8d325a4
Author: Alexey Kuznetsov <ak...@apache.org>
Authored: Wed Oct 28 11:16:39 2015 +0700
Committer: Alexey Kuznetsov <ak...@apache.org>
Committed: Wed Oct 28 11:16:39 2015 +0700
----------------------------------------------------------------------
.../store/jdbc/CacheAbstractJdbcStore.java | 317 ++-
.../cache/store/jdbc/CacheJdbcPojoStore.java | 2001 +-----------------
.../jdbc/CacheJdbcPojoStoreConfiguration.java | 6 +-
.../store/jdbc/CacheJdbcPojoStoreType.java | 272 ---
.../store/jdbc/CacheJdbcPojoStoreTypeField.java | 172 --
.../ignite/cache/store/jdbc/JdbcType.java | 272 +++
.../ignite/cache/store/jdbc/JdbcTypeField.java | 172 ++
.../store/jdbc/CacheJdbcPojoStoreSelfTest.java | 29 +-
.../store/jdbc/CacheJdbcPojoStoreTest.java | 56 +-
.../jdbc/CacheJdbcPortableStoreSelfTest.java | 42 +-
...eJdbcStoreAbstractMultithreadedSelfTest.java | 4 +-
.../jdbc/CacheJdbcStoreAbstractSelfTest.java | 18 +-
12 files changed, 830 insertions(+), 2531 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/2961aaa7/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheAbstractJdbcStore.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheAbstractJdbcStore.java b/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheAbstractJdbcStore.java
index 8b3d44a..576522c 100644
--- a/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheAbstractJdbcStore.java
+++ b/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheAbstractJdbcStore.java
@@ -30,6 +30,7 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.Callable;
@@ -75,6 +76,10 @@ import org.jetbrains.annotations.Nullable;
import static java.sql.Statement.EXECUTE_FAILED;
import static java.sql.Statement.SUCCESS_NO_INFO;
+import static org.apache.ignite.cache.store.jdbc.CacheJdbcPojoStoreConfiguration.DFLT_BATCH_SIZE;
+import static org.apache.ignite.cache.store.jdbc.CacheJdbcPojoStoreConfiguration.DFLT_WRITE_ATTEMPTS;
+import static org.apache.ignite.cache.store.jdbc.CacheJdbcPojoStoreConfiguration.DFLT_PARALLEL_LOAD_CACHE_MINIMUM_THRESHOLD;
+
/**
* Implementation of {@link CacheStore} backed by JDBC.
* <p>
@@ -115,28 +120,16 @@ import static java.sql.Statement.SUCCESS_NO_INFO;
*/
@Deprecated
public abstract class CacheAbstractJdbcStore<K, V> implements CacheStore<K, V>, LifecycleAware {
- /** Max attempt write count. */
- protected static final int MAX_ATTEMPT_WRITE_COUNT = 2;
-
- /** Default batch size for put and remove operations. */
- protected static final int DFLT_BATCH_SIZE = 512;
-
- /** Default batch size for put and remove operations. */
- protected static final int DFLT_PARALLEL_LOAD_CACHE_MINIMUM_THRESHOLD = 512;
-
/** Connection attribute property name. */
protected static final String ATTR_CONN_PROP = "JDBC_STORE_CONNECTION";
- /** Empty column value. */
- protected static final Object[] EMPTY_COLUMN_VALUE = new Object[] { null };
-
/** Auto-injected store session. */
@CacheStoreSessionResource
private CacheStoreSession ses;
/** Auto injected ignite instance. */
@IgniteInstanceResource
- private Ignite ignite;
+ protected Ignite ignite;
/** Auto-injected logger instance. */
@LoggerResource
@@ -152,18 +145,27 @@ public abstract class CacheAbstractJdbcStore<K, V> implements CacheStore<K, V>,
/** Cache with entry mapping description. (cache name, (key id, mapping description)). */
protected volatile Map<String, Map<Object, EntryMapping>> cacheMappings = Collections.emptyMap();
+ /** Map for quick check whether type is POJO or Portable. */
+ private volatile Map<String, Map<String, Boolean>> keepSerializedTypes = new HashMap<>();
+
+ /** Maximum batch size for writeAll and deleteAll operations. */
+ private int batchSz = DFLT_BATCH_SIZE;
+
/** Database dialect. */
protected JdbcDialect dialect;
+ /** Maximum write attempts in case of database error. */
+ private int maxWrtAttempts = DFLT_WRITE_ATTEMPTS;
+
/** Max workers thread count. These threads are responsible for load cache. */
private int maxPoolSz = Runtime.getRuntime().availableProcessors();
- /** Maximum batch size for writeAll and deleteAll operations. */
- private int batchSz = DFLT_BATCH_SIZE;
-
/** Parallel load cache minimum threshold. If {@code 0} then load sequentially. */
private int parallelLoadCacheMinThreshold = DFLT_PARALLEL_LOAD_CACHE_MINIMUM_THRESHOLD;
+ /** Types that store could process. */
+ private JdbcType[] types;
+
/**
* Get field value from object for use as query parameter.
*
@@ -190,26 +192,27 @@ public abstract class CacheAbstractJdbcStore<K, V> implements CacheStore<K, V>,
* @throws CacheLoaderException If failed to construct cache object.
*/
protected abstract <R> R buildObject(@Nullable String cacheName, String typeName,
- Collection<CacheTypeFieldMetadata> fields, Map<String, Integer> loadColIdxs, ResultSet rs)
+ JdbcTypeField[] fields, Map<String, Integer> loadColIdxs, ResultSet rs)
throws CacheLoaderException;
/**
- * Extract key type id from key object.
+ * Calculate type ID for object.
*
- * @param key Key object.
- * @return Key type id.
- * @throws CacheException If failed to get type key id from object.
+ * @param obj Object to calculate type ID for.
+ * @return Type ID.
+ * @throws CacheException If failed to calculate type ID for given object.
*/
- protected abstract Object keyTypeId(Object key) throws CacheException;
+ protected abstract Object typeIdForObject(Object obj) throws CacheException;
/**
- * Extract key type id from key class name.
+ * Calculate type ID for given type name.
*
- * @param type String description of key type.
- * @return Key type id.
- * @throws CacheException If failed to get type key id from object.
+ * @param keepSerialized If {@code true} then calculate type ID for portable object otherwise for POJO.
+ * @param typeName String description of type name.
+ * @return Type ID.
+ * @throws CacheException If failed to get type ID for given type name.
*/
- protected abstract Object keyTypeId(String type) throws CacheException;
+ protected abstract Object typeIdForTypeName(boolean keepSerialized, String typeName) throws CacheException;
/**
* Prepare internal store specific builders for provided types metadata.
@@ -218,7 +221,7 @@ public abstract class CacheAbstractJdbcStore<K, V> implements CacheStore<K, V>,
* @param types Collection of types.
* @throws CacheException If failed to prepare internal builders for types.
*/
- protected abstract void prepareBuilders(@Nullable String cacheName, Collection<CacheTypeMetadata> types)
+ protected abstract void prepareBuilders(@Nullable String cacheName, Collection<JdbcType> types)
throws CacheException;
/**
@@ -541,35 +544,35 @@ public abstract class CacheAbstractJdbcStore<K, V> implements CacheStore<K, V>,
* @param fields Fields descriptors.
* @throws CacheException If failed to check type metadata.
*/
- private static void checkMapping(@Nullable String cacheName, String clsName,
- Collection<CacheTypeFieldMetadata> fields) throws CacheException {
+ private void checkMapping(@Nullable String cacheName, String clsName,
+ JdbcTypeField[] fields) throws CacheException {
try {
Class<?> cls = Class.forName(clsName);
if (simpleType(cls)) {
- if (fields.size() != 1)
+ if (fields.length != 1)
throw new CacheException("More than one field for simple type [cache name=" + cacheName
+ ", type=" + clsName + " ]");
- CacheTypeFieldMetadata field = F.first(fields);
+ JdbcTypeField field = fields[0];
- if (field.getDatabaseName() == null)
+ if (field.getDatabaseFieldName() == null)
throw new CacheException("Missing database name in mapping description [cache name=" + cacheName
+ ", type=" + clsName + " ]");
- field.setJavaType(cls);
+ field.setJavaFieldType(cls);
}
else
- for (CacheTypeFieldMetadata field : fields) {
- if (field.getDatabaseName() == null)
+ for (JdbcTypeField field : fields) {
+ if (field.getDatabaseFieldName() == null)
throw new CacheException("Missing database name in mapping description [cache name=" + cacheName
+ ", type=" + clsName + " ]");
- if (field.getJavaName() == null)
+ if (field.getJavaFieldName() == null)
throw new CacheException("Missing field name in mapping description [cache name=" + cacheName
+ ", type=" + clsName + " ]");
- if (field.getJavaType() == null)
+ if (field.getJavaFieldType() == null)
throw new CacheException("Missing field type in mapping description [cache name=" + cacheName
+ ", type=" + clsName + " ]");
}
@@ -580,6 +583,50 @@ public abstract class CacheAbstractJdbcStore<K, V> implements CacheStore<K, V>,
}
/**
+ * For backward compatibility translate old field type descriptors to new format.
+ *
+ * @param oldFields Fields in old format.
+ * @return Fields in new format.
+ */
+ @Deprecated
+ private JdbcTypeField[] translateFields(Collection<CacheTypeFieldMetadata> oldFields) {
+ JdbcTypeField[] newFields = new JdbcTypeField[oldFields.size()];
+
+ int idx = 0;
+
+ for (CacheTypeFieldMetadata oldField : oldFields) {
+ newFields[idx] = new JdbcTypeField(oldField.getDatabaseType(), oldField.getDatabaseName(),
+ oldField.getJavaType(), oldField.getJavaName());
+
+ idx++;
+ }
+
+ return newFields;
+ }
+
+ /**
+ * Checks for POJO/portable format.
+ *
+ * @param cacheName Cache name to get types settings.
+ * @param typeName Type name to check for POJO/portable format.
+ * @return {@code true} If portable format configured.
+ * @throws CacheException In case of error.
+ */
+ protected boolean isKeepSerialized(String cacheName, String typeName) {
+ Map<String, Boolean> cacheTypes = keepSerializedTypes.get(cacheName);
+
+ if (cacheTypes == null)
+ throw new CacheException("Failed to find types metadata for cache: " + cacheName);
+
+ Boolean keepSerialized = cacheTypes.get(typeName);
+
+ if (keepSerialized == null)
+ throw new CacheException("Failed to find type metadata for type: " + typeName);
+
+ return keepSerialized;
+ }
+
+ /**
* @param cacheName Cache name to check mappings for.
* @return Type mappings for specified cache name.
* @throws CacheException If failed to initialize cache mappings.
@@ -598,32 +645,80 @@ public abstract class CacheAbstractJdbcStore<K, V> implements CacheStore<K, V>,
if (entryMappings != null)
return entryMappings;
- CacheConfiguration ccfg = ignite().cache(cacheName).getConfiguration(CacheConfiguration.class);
+ // If no types configured, check CacheTypeMetadata for backward compatibility.
+ if (types == null) {
+ CacheConfiguration ccfg = ignite().cache(cacheName).getConfiguration(CacheConfiguration.class);
+
+ Collection<CacheTypeMetadata> oldTypes = ccfg.getTypeMetadata();
+
+ types = new JdbcType[oldTypes.size()];
+
+ int idx = 0;
- Collection<CacheTypeMetadata> types = ccfg.getTypeMetadata();
+ for (CacheTypeMetadata oldType : oldTypes) {
+ JdbcType newType = new JdbcType();
- entryMappings = U.newHashMap(types.size());
+ newType.setCacheName(cacheName);
- for (CacheTypeMetadata type : types) {
- Object keyTypeId = keyTypeId(type.getKeyType());
+ newType.setDatabaseSchema(oldType.getDatabaseSchema());
+ newType.setDatabaseTable(oldType.getDatabaseTable());
- if (entryMappings.containsKey(keyTypeId))
- throw new CacheException("Key type must be unique in type metadata [cache name=" + cacheName +
- ", key type=" + type.getKeyType() + "]");
+ newType.setKeyType(oldType.getKeyType());
+ newType.setKeyFields(translateFields(oldType.getKeyFields()));
- checkMapping(cacheName, type.getKeyType(), type.getKeyFields());
- checkMapping(cacheName, type.getValueType(), type.getValueFields());
+ newType.setValueType(oldType.getValueType());
+ newType.setValueFields(translateFields(oldType.getValueFields()));
- entryMappings.put(keyTypeId(type.getKeyType()), new EntryMapping(cacheName, dialect, type));
+ types[idx] = newType;
+
+ idx++;
+ }
}
- Map<String, Map<Object, EntryMapping>> mappings = new HashMap<>(cacheMappings);
+ List<JdbcType> cacheTypes = new ArrayList<>(types.length);
+
+ for (JdbcType type : types)
+ if ((cacheName != null && cacheName.equals(type.getCacheName())) ||
+ (cacheName == null && type.getCacheName() == null))
+ cacheTypes.add(type);
+
+ entryMappings = U.newHashMap(cacheTypes.size());
+
+ if (!cacheTypes.isEmpty()) {
+ Map<String, Boolean> tk = new HashMap<>(cacheTypes.size() * 2);
- mappings.put(cacheName, entryMappings);
+ for (JdbcType type : cacheTypes) {
+ boolean keepSerialized = type.isKeepSerialized();
- prepareBuilders(cacheName, types);
+ String keyType = type.getKeyType();
+ String valType = type.getValueType();
- cacheMappings = mappings;
+ tk.put(keyType, keepSerialized);
+ tk.put(valType, keepSerialized);
+
+ Object keyTypeId = typeIdForTypeName(keepSerialized, keyType);
+
+ if (entryMappings.containsKey(keyTypeId))
+ throw new CacheException("Key type must be unique in type metadata [cache name=" + cacheName +
+ ", key type=" + keyType + "]");
+
+ if (!keepSerialized) {
+ checkMapping(cacheName, keyType, type.getKeyFields());
+ checkMapping(cacheName, valType, type.getValueFields());
+ }
+
+ entryMappings.put(keyTypeId, new EntryMapping(cacheName, dialect, type));
+ }
+
+ keepSerializedTypes.put(cacheName, tk);
+
+ Map<String, Map<Object, EntryMapping>> mappings = new HashMap<>(cacheMappings);
+ mappings.put(cacheName, entryMappings);
+
+ prepareBuilders(cacheName, cacheTypes);
+
+ cacheMappings = mappings;
+ }
return entryMappings;
}
@@ -646,7 +741,7 @@ public abstract class CacheAbstractJdbcStore<K, V> implements CacheStore<K, V>,
String maskedCacheName = U.maskName(cacheName);
throw new CacheException("Failed to find mapping description [key=" + key +
- ", cache=" + maskedCacheName + "]. Please configure CacheTypeMetadata to associate '" + maskedCacheName + "' with JdbcPojoStore.");
+ ", cache=" + maskedCacheName + "]. Please configure JdbcType to associate '" + maskedCacheName + "' with JdbcPojoStore.");
}
return em;
@@ -671,12 +766,16 @@ public abstract class CacheAbstractJdbcStore<K, V> implements CacheStore<K, V>,
if (log.isDebugEnabled())
log.debug("Start loading entries from db using user queries from arguments...");
+ // We must build cache mappings first.
+ cacheMappings(cacheName);
+
for (int i = 0; i < args.length; i += 2) {
String keyType = args[i].toString();
String selQry = args[i + 1].toString();
- EntryMapping em = entryMapping(cacheName, keyTypeId(keyType), keyType);
+ EntryMapping em = entryMapping(cacheName, typeIdForTypeName(isKeepSerialized(cacheName, keyType),
+ keyType), keyType);
futs.add(pool.submit(new LoadCacheCustomQueryWorker<>(em, selQry, clo)));
}
@@ -762,7 +861,7 @@ public abstract class CacheAbstractJdbcStore<K, V> implements CacheStore<K, V>,
@Nullable @Override public V load(K key) throws CacheLoaderException {
assert key != null;
- EntryMapping em = entryMapping(session().cacheName(), keyTypeId(key), key);
+ EntryMapping em = entryMapping(session().cacheName(), typeIdForObject(key), key);
if (log.isDebugEnabled())
log.debug("Load value from db [table= " + em.fullTableName() + ", key=" + key + "]");
@@ -810,7 +909,7 @@ public abstract class CacheAbstractJdbcStore<K, V> implements CacheStore<K, V>,
Map<K, V> res = new HashMap<>();
for (K key : keys) {
- Object keyTypeId = keyTypeId(key);
+ Object keyTypeId = typeIdForObject(key);
EntryMapping em = entryMapping(cacheName, keyTypeId, key);
@@ -850,7 +949,7 @@ public abstract class CacheAbstractJdbcStore<K, V> implements CacheStore<K, V>,
try {
CacheWriterException we = null;
- for (int attempt = 0; attempt < MAX_ATTEMPT_WRITE_COUNT; attempt++) {
+ for (int attempt = 0; attempt < maxWrtAttempts; attempt++) {
int paramIdx = fillValueParameters(updStmt, 1, em, entry.getValue());
fillKeyParameters(updStmt, paramIdx, em, entry.getKey());
@@ -919,7 +1018,7 @@ public abstract class CacheAbstractJdbcStore<K, V> implements CacheStore<K, V>,
K key = entry.getKey();
- EntryMapping em = entryMapping(session().cacheName(), keyTypeId(key), key);
+ EntryMapping em = entryMapping(session().cacheName(), typeIdForObject(key), key);
if (log.isDebugEnabled())
log.debug("Start write entry to database [table=" + em.fullTableName() + ", entry=" + entry + "]");
@@ -1008,7 +1107,7 @@ public abstract class CacheAbstractJdbcStore<K, V> implements CacheStore<K, V>,
for (Cache.Entry<? extends K, ? extends V> entry : entries) {
K key = entry.getKey();
- Object keyTypeId = keyTypeId(key);
+ Object keyTypeId = typeIdForObject(key);
em = entryMapping(cacheName, keyTypeId, key);
@@ -1077,7 +1176,7 @@ public abstract class CacheAbstractJdbcStore<K, V> implements CacheStore<K, V>,
for (Cache.Entry<? extends K, ? extends V> entry : entries) {
K key = entry.getKey();
- Object keyTypeId = keyTypeId(key);
+ Object keyTypeId = typeIdForObject(key);
EntryMapping em = entryMapping(cacheName, keyTypeId, key);
@@ -1115,7 +1214,7 @@ public abstract class CacheAbstractJdbcStore<K, V> implements CacheStore<K, V>,
@Override public void delete(Object key) throws CacheWriterException {
assert key != null;
- EntryMapping em = entryMapping(session().cacheName(), keyTypeId(key), key);
+ EntryMapping em = entryMapping(session().cacheName(), typeIdForObject(key), key);
if (log.isDebugEnabled())
log.debug("Remove value from db [table=" + em.fullTableName() + ", key=" + key + "]");
@@ -1219,7 +1318,7 @@ public abstract class CacheAbstractJdbcStore<K, V> implements CacheStore<K, V>,
int fromIdx = 0, prepared = 0;
for (Object key : keys) {
- Object keyTypeId = keyTypeId(key);
+ Object keyTypeId = typeIdForObject(key);
em = entryMapping(cacheName, keyTypeId, key);
@@ -1285,12 +1384,12 @@ public abstract class CacheAbstractJdbcStore<K, V> implements CacheStore<K, V>,
* @param fieldVal Field value.
* @throws CacheException If failed to set statement parameter.
*/
- protected void fillParameter(PreparedStatement stmt, int i, CacheTypeFieldMetadata field, @Nullable Object fieldVal)
+ protected void fillParameter(PreparedStatement stmt, int i, JdbcTypeField field, @Nullable Object fieldVal)
throws CacheException {
try {
if (fieldVal != null) {
- if (field.getJavaType() == UUID.class) {
- switch (field.getDatabaseType()) {
+ if (field.getJavaFieldType() == UUID.class) {
+ switch (field.getDatabaseFieldType()) {
case Types.BINARY:
fieldVal = U.uuidToBytes((UUID)fieldVal);
@@ -1306,10 +1405,10 @@ public abstract class CacheAbstractJdbcStore<K, V> implements CacheStore<K, V>,
stmt.setObject(i, fieldVal);
}
else
- stmt.setNull(i, field.getDatabaseType());
+ stmt.setNull(i, field.getDatabaseFieldType());
}
catch (SQLException e) {
- throw new CacheException("Failed to set statement parameter name: " + field.getDatabaseName(), e);
+ throw new CacheException("Failed to set statement parameter name: " + field.getDatabaseFieldName(), e);
}
}
@@ -1323,8 +1422,8 @@ public abstract class CacheAbstractJdbcStore<K, V> implements CacheStore<K, V>,
*/
protected int fillKeyParameters(PreparedStatement stmt, int idx, EntryMapping em,
Object key) throws CacheException {
- for (CacheTypeFieldMetadata field : em.keyColumns()) {
- Object fieldVal = extractParameter(em.cacheName, em.keyType(), field.getJavaName(), key);
+ for (JdbcTypeField field : em.keyColumns()) {
+ Object fieldVal = extractParameter(em.cacheName, em.keyType(), field.getJavaFieldName(), key);
fillParameter(stmt, idx++, field, fieldVal);
}
@@ -1353,8 +1452,8 @@ public abstract class CacheAbstractJdbcStore<K, V> implements CacheStore<K, V>,
*/
protected int fillValueParameters(PreparedStatement stmt, int idx, EntryMapping em, Object val)
throws CacheWriterException {
- for (CacheTypeFieldMetadata field : em.uniqValFields) {
- Object fieldVal = extractParameter(em.cacheName, em.valueType(), field.getJavaName(), val);
+ for (JdbcTypeField field : em.uniqValFields) {
+ Object fieldVal = extractParameter(em.cacheName, em.valueType(), field.getJavaFieldName(), val);
fillParameter(stmt, idx++, field, fieldVal);
}
@@ -1413,6 +1512,42 @@ public abstract class CacheAbstractJdbcStore<K, V> implements CacheStore<K, V>,
}
/**
+ * Gets maximum number of write attempts in case of database error.
+ *
+ * @return Maximum number of write attempts.
+ */
+ public int getMaximumWriteAttempts() {
+ return maxWrtAttempts;
+ }
+
+ /**
+ * Sets maximum number of write attempts in case of database error.
+ *
+ * @param maxWrtAttempts Number of write attempts.
+ */
+ public void setMaximumWriteAttempts(int maxWrtAttempts) {
+ this.maxWrtAttempts = maxWrtAttempts;
+ }
+
+ /**
+ * Gets types known by store.
+ *
+ * @return Types known by store.
+ */
+ public JdbcType[] getTypes() {
+ return types;
+ }
+
+ /**
+ * Sets store configurations.
+ *
+ * @param types Store should process.
+ */
+ public void setTypes(JdbcType... types) {
+ this.types = types;
+ }
+
+ /**
* Get maximum batch size for delete and delete operations.
*
* @return Maximum batch size.
@@ -1509,10 +1644,10 @@ public abstract class CacheAbstractJdbcStore<K, V> implements CacheStore<K, V>,
private final Map<String, Integer> loadColIdxs;
/** Unique value fields. */
- private final Collection<CacheTypeFieldMetadata> uniqValFields;
+ private final Collection<JdbcTypeField> uniqValFields;
/** Type metadata. */
- private final CacheTypeMetadata typeMeta;
+ private final JdbcType typeMeta;
/** Full table name. */
private final String fullTblName;
@@ -1522,22 +1657,22 @@ public abstract class CacheAbstractJdbcStore<K, V> implements CacheStore<K, V>,
* @param dialect JDBC dialect.
* @param typeMeta Type metadata.
*/
- public EntryMapping(@Nullable String cacheName, JdbcDialect dialect, CacheTypeMetadata typeMeta) {
+ public EntryMapping(@Nullable String cacheName, JdbcDialect dialect, JdbcType typeMeta) {
this.cacheName = cacheName;
this.dialect = dialect;
this.typeMeta = typeMeta;
- Collection<CacheTypeFieldMetadata> keyFields = typeMeta.getKeyFields();
+ JdbcTypeField[] keyFields = typeMeta.getKeyFields();
- Collection<CacheTypeFieldMetadata> valFields = typeMeta.getValueFields();
+ JdbcTypeField[] valFields = typeMeta.getValueFields();
- keyCols = databaseColumns(keyFields);
+ keyCols = databaseColumns(F.asList(keyFields));
- uniqValFields = F.view(valFields, new IgnitePredicate<CacheTypeFieldMetadata>() {
- @Override public boolean apply(CacheTypeFieldMetadata col) {
- return !keyCols.contains(col.getDatabaseName());
+ uniqValFields = F.view(F.asList(valFields), new IgnitePredicate<JdbcTypeField>() {
+ @Override public boolean apply(JdbcTypeField col) {
+ return !keyCols.contains(col.getDatabaseFieldName());
}
});
@@ -1578,16 +1713,16 @@ public abstract class CacheAbstractJdbcStore<K, V> implements CacheStore<K, V>,
}
/**
- * Extract database column names from {@link CacheTypeFieldMetadata}.
+ * Extract database column names from {@link JdbcTypeField}.
*
- * @param dsc collection of {@link CacheTypeFieldMetadata}.
+ * @param dsc collection of {@link JdbcTypeField}.
* @return Collection with database column names.
*/
- private static Collection<String> databaseColumns(Collection<CacheTypeFieldMetadata> dsc) {
- return F.transform(dsc, new C1<CacheTypeFieldMetadata, String>() {
+ private static Collection<String> databaseColumns(Collection<JdbcTypeField> dsc) {
+ return F.transform(dsc, new C1<JdbcTypeField, String>() {
/** {@inheritDoc} */
- @Override public String apply(CacheTypeFieldMetadata col) {
- return col.getDatabaseName();
+ @Override public String apply(JdbcTypeField col) {
+ return col.getDatabaseFieldName();
}
});
}
@@ -1640,7 +1775,7 @@ public abstract class CacheAbstractJdbcStore<K, V> implements CacheStore<K, V>,
*
* @return Key columns.
*/
- protected Collection<CacheTypeFieldMetadata> keyColumns() {
+ protected JdbcTypeField[] keyColumns() {
return typeMeta.getKeyFields();
}
@@ -1649,7 +1784,7 @@ public abstract class CacheAbstractJdbcStore<K, V> implements CacheStore<K, V>,
*
* @return Value columns.
*/
- protected Collection<CacheTypeFieldMetadata> valueColumns() {
+ protected JdbcTypeField[] valueColumns() {
return typeMeta.getValueFields();
}
@@ -1799,8 +1934,8 @@ public abstract class CacheAbstractJdbcStore<K, V> implements CacheStore<K, V>,
int idx = 1;
for (Object key : keys)
- for (CacheTypeFieldMetadata field : em.keyColumns()) {
- Object fieldVal = extractParameter(em.cacheName, em.keyType(), field.getJavaName(), key);
+ for (JdbcTypeField field : em.keyColumns()) {
+ Object fieldVal = extractParameter(em.cacheName, em.keyType(), field.getJavaFieldName(), key);
fillParameter(stmt, idx++, field, fieldVal);
}
[4/4] ignite git commit: IGNITE-1753 WIP: added support for simple
types in case of portables.
Posted by ak...@apache.org.
IGNITE-1753 WIP: added support for simple types in case of portables.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/852e54c0
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/852e54c0
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/852e54c0
Branch: refs/heads/ignite-1753-1282
Commit: 852e54c068915ecd9df86c850048afb7797a0a9e
Parents: 2961aaa
Author: Alexey Kuznetsov <ak...@apache.org>
Authored: Wed Oct 28 15:19:12 2015 +0700
Committer: Alexey Kuznetsov <ak...@apache.org>
Committed: Wed Oct 28 15:19:12 2015 +0700
----------------------------------------------------------------------
.../store/jdbc/CacheAbstractJdbcStore.java | 21 ++++++++
.../cache/store/jdbc/CacheJdbcPojoStore.java | 57 ++++++++++++--------
2 files changed, 55 insertions(+), 23 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/852e54c0/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheAbstractJdbcStore.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheAbstractJdbcStore.java b/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheAbstractJdbcStore.java
index 576522c..4a81135 100644
--- a/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheAbstractJdbcStore.java
+++ b/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheAbstractJdbcStore.java
@@ -30,6 +30,7 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.UUID;
@@ -123,6 +124,26 @@ public abstract class CacheAbstractJdbcStore<K, V> implements CacheStore<K, V>,
/** Connection attribute property name. */
protected static final String ATTR_CONN_PROP = "JDBC_STORE_CONNECTION";
+ /** Simple types names. */
+ protected static Collection<String> SIMPLE_TYPES = new HashSet<>();
+
+ static {
+ SIMPLE_TYPES.add("java.math.BigDecimal");
+ SIMPLE_TYPES.add("java.lang.Boolean");
+ SIMPLE_TYPES.add("java.lang.Byte");
+ SIMPLE_TYPES.add("java.lang.Character");
+ SIMPLE_TYPES.add("java.lang.Double");
+ SIMPLE_TYPES.add("java.util.Date");
+ SIMPLE_TYPES.add("java.sql.Date");
+ SIMPLE_TYPES.add("java.lang.Float");
+ SIMPLE_TYPES.add("java.lang.Integer");
+ SIMPLE_TYPES.add("java.lang.Long");
+ SIMPLE_TYPES.add("java.lang.Short");
+ SIMPLE_TYPES.add("java.lang.String");
+ SIMPLE_TYPES.add("java.sql.Timestamp");
+ SIMPLE_TYPES.add("java.util.UUID");
+ }
+
/** Auto-injected store session. */
@CacheStoreSessionResource
private CacheStoreSession ses;
http://git-wip-us.apache.org/repos/asf/ignite/blob/852e54c0/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 1724fa0..e3be95b 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
@@ -32,6 +32,7 @@ import org.apache.ignite.IgnitePortables;
import org.apache.ignite.cache.store.CacheStore;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lang.IgniteBiTuple;
import org.apache.ignite.portable.PortableBuilder;
import org.apache.ignite.portable.PortableObject;
import org.jetbrains.annotations.Nullable;
@@ -47,7 +48,7 @@ public class CacheJdbcPojoStore<K, V> extends CacheAbstractJdbcStore<K, V> {
private volatile Map<String, Map<String, PojoMethodsCache>> pojoMethods = Collections.emptyMap();
/** Portables builders cache. */
- private volatile Map<String, Map<String, Integer>> portableTypeIds = Collections.emptyMap();
+ private volatile Map<String, Map<String, IgniteBiTuple<Boolean, Integer>>> portableTypeIds = Collections.emptyMap();
/**
* Get field value from object for use as query parameter.
@@ -221,40 +222,50 @@ public class CacheJdbcPojoStore<K, V> extends CacheAbstractJdbcStore<K, V> {
* @return Constructed portable object.
* @throws CacheLoaderException If failed to construct portable object.
*/
- protected PortableObject buildPortableObject(String cacheName, String typeName, JdbcTypeField[] fields,
+ protected Object buildPortableObject(String cacheName, String typeName, JdbcTypeField[] fields,
Map<String, Integer> loadColIdxs, ResultSet rs) throws CacheException {
- Map<String, Integer> cacheTypeIds = portableTypeIds.get(cacheName);
+ Map<String, IgniteBiTuple<Boolean, Integer>> cacheTypeIds = portableTypeIds.get(cacheName);
if (cacheTypeIds == null)
throw new CacheLoaderException("Failed to find portable types IDs for cache: " + cacheName);
- Integer typeId = cacheTypeIds.get(typeName);
+ IgniteBiTuple<Boolean, Integer> tuple = cacheTypeIds.get(typeName);
- if (typeId == null)
+ if (tuple == null)
throw new CacheLoaderException("Failed to find portable type ID for type: " + typeName);
- PortableBuilder builder = ignite.portables().builder(typeId);
+ try {
+ if (tuple.get1()) {
+ Object simple = null;
- if (builder == null)
- throw new CacheException("Failed to find portable builder for type: " + typeName);
+ if (fields.length > 0) {
+ JdbcTypeField field = fields[0];
- try {
- int hashCode = 1;
+ Integer colIdx = loadColIdxs.get(field.getDatabaseFieldName());
- for (JdbcTypeField field : fields) {
- Class<?> type = field.getJavaFieldType();
+ simple = getColumnValue(rs, colIdx, field.getJavaFieldType());
+ }
+
+ return simple;
+ }
+ else {
+ PortableBuilder builder = ignite.portables().builder(tuple.get2());
- Integer colIdx = loadColIdxs.get(field.getDatabaseFieldName());
+ int hashCode = 1;
- Object colVal = getColumnValue(rs, colIdx, type);
+ for (JdbcTypeField field : fields) {
+ Integer colIdx = loadColIdxs.get(field.getDatabaseFieldName());
- if (colVal != null)
- hashCode = 31 * hashCode + colVal.hashCode();
+ Object colVal = getColumnValue(rs, colIdx, field.getJavaFieldType());
- builder.setField(field.getJavaFieldName(), colVal);
- }
+ if (colVal != null)
+ hashCode = 31 * hashCode + colVal.hashCode();
- return builder.hashCode(hashCode).build();
+ builder.setField(field.getJavaFieldName(), colVal);
+ }
+
+ return builder.hashCode(hashCode).build();
+ }
}
catch (SQLException e) {
throw new CacheException("Failed to read portable object", e);
@@ -354,7 +365,7 @@ public class CacheJdbcPojoStore<K, V> extends CacheAbstractJdbcStore<K, V> {
*/
private void preparePortableBuilders(@Nullable String cacheName, Collection<JdbcType> types)
throws CacheException {
- Map<String, Integer> typeIds = U.newHashMap(types.size() * 2);
+ Map<String, IgniteBiTuple<Boolean, Integer>> typeIds = U.newHashMap(types.size() * 2);
for (JdbcType type : types) {
if (type.isKeepSerialized()) {
@@ -363,15 +374,15 @@ public class CacheJdbcPojoStore<K, V> extends CacheAbstractJdbcStore<K, V> {
IgnitePortables portables = ignite.portables();
String keyType = type.getKeyType();
- typeIds.put(keyType, portables.typeId(keyType));
+ typeIds.put(keyType, new IgniteBiTuple<>(SIMPLE_TYPES.contains(keyType), portables.typeId(keyType)));
String valType = type.getValueType();
- typeIds.put(valType, portables.typeId(valType));
+ typeIds.put(valType, new IgniteBiTuple<>(SIMPLE_TYPES.contains(valType), portables.typeId(valType)));
}
}
if (!typeIds.isEmpty()) {
- Map<String, Map<String, Integer>> newBuilders = new HashMap<>(portableTypeIds);
+ Map<String, Map<String, IgniteBiTuple<Boolean, Integer>>> newBuilders = new HashMap<>(portableTypeIds);
newBuilders.put(cacheName, typeIds);