modules/cassandra/store/src/main/java/org/apache/ignite/cache/store/cassandra/persistence/KeyValuePersistenceSettings.java - ignite - Git at Google

 /*
  * 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.cassandra.persistence;

 import java.io.BufferedReader;
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.InputStreamReader;
 import java.io.Serializable;
 import java.io.StringReader;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Set;
 import javax.xml.parsers.DocumentBuilder;
 import javax.xml.parsers.DocumentBuilderFactory;
 import org.apache.ignite.IgniteException;
 import org.apache.ignite.cache.store.cassandra.common.CassandraHelper;
 import org.apache.ignite.cache.store.cassandra.common.SystemHelper;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.springframework.core.io.Resource;
 import org.w3c.dom.Document;
 import org.w3c.dom.Element;
 import org.w3c.dom.Node;
 import org.w3c.dom.NodeList;
 import org.xml.sax.InputSource;

 /**
  * Stores persistence settings for Ignite cache key and value
  */
 public class KeyValuePersistenceSettings implements Serializable {
     /**
      * Default Cassandra keyspace options which should be used to create new keyspace.
      * <ul>
      * <li> <b>SimpleStrategy</b> for replication work well for single data center Cassandra cluster.<br/>
      *      If your Cassandra cluster deployed across multiple data centers it's better to use <b>NetworkTopologyStrategy</b>.
      * </li>
      * <li> Three replicas will be created for each data block. </li>
      * <li> Setting DURABLE_WRITES to true specifies that all data should be written to commit log. </li>
      * </ul>
      */
     private static final String DFLT_KEYSPACE_OPTIONS = "replication = {'class' : 'SimpleStrategy', " +
             "'replication_factor' : 3} and durable_writes = true";

     /** Xml attribute specifying Cassandra keyspace to use. */
     private static final String KEYSPACE_ATTR = "keyspace";

     /** Xml attribute specifying Cassandra table to use. */
     private static final String TABLE_ATTR = "table";

     /** Xml attribute specifying ttl (time to leave) for rows inserted in Cassandra. */
     private static final String TTL_ATTR = "ttl";

     /** Root xml element containing persistence settings specification. */
     private static final String PERSISTENCE_NODE = "persistence";

     /** Xml element specifying Cassandra keyspace options. */
     private static final String KEYSPACE_OPTIONS_NODE = "keyspaceOptions";

     /** Xml element specifying Cassandra table options. */
     private static final String TABLE_OPTIONS_NODE = "tableOptions";

     /** Xml element specifying Ignite cache key persistence settings. */
     private static final String KEY_PERSISTENCE_NODE = "keyPersistence";

     /** Xml element specifying Ignite cache value persistence settings. */
     private static final String VALUE_PERSISTENCE_NODE = "valuePersistence";

     /** TTL (time to leave) for rows inserted into Cassandra table {@link <a href="https://docs.datastax.com/en/cql/3.1/cql/cql_using/use_expire_c.html">Expiring data</a>}. */
     private Integer ttl;

     /** Cassandra keyspace (analog of tablespace in relational databases). */
     private String keyspace;

     /** Cassandra table. */
     private String tbl;

     /** Cassandra table creation options {@link <a href="https://docs.datastax.com/en/cql/3.0/cql/cql_reference/create_table_r.html">CREATE TABLE</a>}. */
     private String tblOptions;

     /** Cassandra keyspace creation options {@link <a href="https://docs.datastax.com/en/cql/3.0/cql/cql_reference/create_keyspace_r.html">CREATE KEYSPACE</a>}. */
     private String keyspaceOptions = DFLT_KEYSPACE_OPTIONS;

     /** Persistence settings for Ignite cache keys. */
     private KeyPersistenceSettings keyPersistenceSettings;

     /** Persistence settings for Ignite cache values. */
     private ValuePersistenceSettings valPersistenceSettings;

     /** List of Cassandra table columns */
     private List<String> tableColumns;

     /**
      * Constructs Ignite cache key/value persistence settings.
      *
      * @param settings string containing xml with persistence settings for Ignite cache key/value
      */
     public KeyValuePersistenceSettings(String settings) {
         init(settings);
     }

     /**
      * Constructs Ignite cache key/value persistence settings.
      *
      * @param settingsFile xml file with persistence settings for Ignite cache key/value
      */
     public KeyValuePersistenceSettings(File settingsFile) {
         InputStream in;

         try {
             in = new FileInputStream(settingsFile);
         }
         catch (IOException e) {
             throw new IgniteException("Failed to get input stream for Cassandra persistence settings file: " +
                     settingsFile.getAbsolutePath(), e);
         }

         init(loadSettings(in));
     }

     /**
      * Constructs Ignite cache key/value persistence settings.
      *
      * @param settingsRsrc resource containing xml with persistence settings for Ignite cache key/value
      */
     public KeyValuePersistenceSettings(Resource settingsRsrc) {
         InputStream in;

         try {
             in = settingsRsrc.getInputStream();
         }
         catch (IOException e) {
             throw new IgniteException("Failed to get input stream for Cassandra persistence settings resource: " + settingsRsrc, e);
         }

         init(loadSettings(in));
     }

     /**
      * Returns ttl to use for while inserting new rows into Cassandra table.
      *
      * @return ttl
      */
     public Integer getTTL() {
         return ttl;
     }

     /**
      * Returns Cassandra keyspace to use.
      *
      * @return keyspace.
      */
     public String getKeyspace() {
         return keyspace;
     }

     /**
      * Returns Cassandra table to use.
      *
      * @return table.
      */
     public String getTable() {
         return tbl;
     }

     /**
      * Returns persistence settings for Ignite cache keys.
      *
      * @return keys persistence settings.
      */
     public KeyPersistenceSettings getKeyPersistenceSettings() {
         return keyPersistenceSettings;
     }

     /**
      * Returns persistence settings for Ignite cache values.
      *
      * @return values persistence settings.
      */
     public ValuePersistenceSettings getValuePersistenceSettings() {
         return valPersistenceSettings;
     }

     /**
      * Returns list of POJO fields to be mapped to Cassandra table columns.
      *
      * @return POJO fields list.
      */
     public List<PojoField> getFields() {
         List<PojoField> fields = new LinkedList<>();

         for (PojoField field : keyPersistenceSettings.getFields())
             fields.add(field);

         for (PojoField field : valPersistenceSettings.getFields())
             fields.add(field);

         return fields;
     }

     /**
      * Returns list of Ignite cache key POJO fields to be mapped to Cassandra table columns.
      *
      * @return POJO fields list.
      */
     public List<PojoKeyField> getKeyFields() {
         return keyPersistenceSettings.getFields();
     }

     /**
      * Returns list of Ignite cache value POJO fields to be mapped to Cassandra table columns.
      *
      * @return POJO fields list.
      */
     public List<PojoValueField> getValueFields() {
         return valPersistenceSettings.getFields();
     }

     /**
      * Returns DDL statement to create Cassandra keyspace.
      *
      * @return Keyspace DDL statement.
      */
     public String getKeyspaceDDLStatement() {
         StringBuilder builder = new StringBuilder();
         builder.append("create keyspace if not exists \"").append(keyspace).append("\"");

         if (keyspaceOptions != null) {
             if (!keyspaceOptions.trim().toLowerCase().startsWith("with"))
                 builder.append("\nwith");

             builder.append(" ").append(keyspaceOptions);
         }

         String statement = builder.toString().trim().replaceAll(" +", " ");

         return statement.endsWith(";") ? statement : statement + ";";
     }

     /**
      * Returns column names for Cassandra table.
      *
      * @return Column names.
      */
     public List<String> getTableColumns() {
         return tableColumns;
     }

     /**
      * Returns DDL statement to create Cassandra table.
      *
      * @param table Table name.
      * @return Table DDL statement.
      */
     public String getTableDDLStatement(String table) {
         if (table == null || table.trim().isEmpty())
             throw new IllegalArgumentException("Table name should be specified");

         String keyColumnsDDL = keyPersistenceSettings.getTableColumnsDDL();
         String valColumnsDDL = valPersistenceSettings.getTableColumnsDDL(new HashSet<>(keyPersistenceSettings.getTableColumns()));

         String colsDDL = keyColumnsDDL;

         if (valColumnsDDL != null && !valColumnsDDL.trim().isEmpty())
             colsDDL += ",\n" + valColumnsDDL;

         String primaryKeyDDL = keyPersistenceSettings.getPrimaryKeyDDL();

         String clusteringDDL = keyPersistenceSettings.getClusteringDDL();

         String optionsDDL = tblOptions != null && !tblOptions.trim().isEmpty() ? tblOptions.trim() : "";

         if (clusteringDDL != null && !clusteringDDL.isEmpty())
             optionsDDL = optionsDDL.isEmpty() ? clusteringDDL : optionsDDL + " and " + clusteringDDL;

         if (!optionsDDL.trim().isEmpty())
             optionsDDL = optionsDDL.trim().toLowerCase().startsWith("with") ? optionsDDL.trim() : "with " + optionsDDL.trim();

         StringBuilder builder = new StringBuilder();

         builder.append("create table if not exists \"").append(keyspace).append("\".\"").append(table).append("\"");
         builder.append("\n(\n").append(colsDDL).append(",\n").append(primaryKeyDDL).append("\n)");

         if (!optionsDDL.isEmpty())
             builder.append(" \n").append(optionsDDL);

         String tblDDL = builder.toString().trim().replaceAll(" +", " ");

         return tblDDL.endsWith(";") ? tblDDL : tblDDL + ";";
     }

     /**
      * Returns DDL statements to create Cassandra table secondary indexes.
      *
      * @param table Table name.
      * @return DDL statements to create secondary indexes.
      */
     public List<String> getIndexDDLStatements(String table) {
         List<String> idxDDLs = new LinkedList<>();

         Set<String> keyCols = new HashSet<>(keyPersistenceSettings.getTableColumns());

         List<PojoValueField> fields = valPersistenceSettings.getFields();

         for (PojoField field : fields) {
             if (!keyCols.contains(field.getColumn()) && ((PojoValueField)field).isIndexed())
                 idxDDLs.add(((PojoValueField)field).getIndexDDL(keyspace, table));
         }

         return idxDDLs;
     }

     /**
      * Loads Ignite cache persistence settings from resource.
      *
      * @param in Input stream.
      * @return String containing xml with Ignite cache persistence settings.
      */
     private String loadSettings(InputStream in) {
         StringBuilder settings = new StringBuilder();
         BufferedReader reader = null;

         try {
             reader = new BufferedReader(new InputStreamReader(in));

             String line = reader.readLine();

             while (line != null) {
                 if (settings.length() != 0)
                     settings.append(SystemHelper.LINE_SEPARATOR);

                 settings.append(line);

                 line = reader.readLine();
             }
         }
         catch (Throwable e) {
             throw new IgniteException("Failed to read input stream for Cassandra persistence settings", e);
         }
         finally {
             U.closeQuiet(reader);
             U.closeQuiet(in);
         }

         return settings.toString();
     }

     /**
      * @param elem Element with data.
      * @param attr Attribute name.
      * @return Numeric value for specified attribute.
      */
     private int extractIntAttribute(Element elem, String attr) {
         String val = elem.getAttribute(attr).trim();

         try {
             return Integer.parseInt(val);
         }
         catch (NumberFormatException ignored) {
             throw new IllegalArgumentException("Incorrect value '" + val + "' specified for '" + attr + "' attribute");
         }
     }

     /**
      * Initializes persistence settings from XML string.
      *
      * @param settings XML string containing Ignite cache persistence settings configuration.
      */
     @SuppressWarnings("IfCanBeSwitch")
     private void init(String settings) {
         Document doc;

         try {
             DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance();
             DocumentBuilder builder = factory.newDocumentBuilder();
             doc = builder.parse(new InputSource(new StringReader(settings)));
         }
         catch (Throwable e) {
             throw new IllegalArgumentException("Failed to parse persistence settings:" +
                 SystemHelper.LINE_SEPARATOR + settings, e);
         }

         Element root = doc.getDocumentElement();

         if (!PERSISTENCE_NODE.equals(root.getNodeName())) {
             throw new IllegalArgumentException("Incorrect persistence settings specified. " +
                 "Root XML element should be 'persistence'");
         }

         if (!root.hasAttribute(KEYSPACE_ATTR)) {
             throw new IllegalArgumentException("Incorrect persistence settings '" + KEYSPACE_ATTR +
                 "' attribute should be specified");
         }

         keyspace = root.getAttribute(KEYSPACE_ATTR).trim();
         tbl = root.hasAttribute(TABLE_ATTR) ? root.getAttribute(TABLE_ATTR).trim() : null;

         if (root.hasAttribute(TTL_ATTR))
             ttl = extractIntAttribute(root, TTL_ATTR);

         if (!root.hasChildNodes()) {
             throw new IllegalArgumentException("Incorrect Cassandra persistence settings specification, " +
                 "there are no key and value persistence settings specified");
         }

         NodeList children = root.getChildNodes();
         int cnt = children.getLength();

         for (int i = 0; i < cnt; i++) {
             Node node = children.item(i);

             if (node.getNodeType() != Node.ELEMENT_NODE)
                 continue;

             Element el = (Element)node;
             String nodeName = el.getNodeName();

             if (nodeName.equals(TABLE_OPTIONS_NODE)) {
                 tblOptions = el.getTextContent();
                 tblOptions = tblOptions.replace("\n", " ").replace("\r", "").replace("\t", " ");
             }
             else if (nodeName.equals(KEYSPACE_OPTIONS_NODE)) {
                 keyspaceOptions = el.getTextContent();
                 keyspaceOptions = keyspaceOptions.replace("\n", " ").replace("\r", "").replace("\t", " ");
             }
             else if (nodeName.equals(KEY_PERSISTENCE_NODE))
                 keyPersistenceSettings = new KeyPersistenceSettings(el);
             else if (nodeName.equals(VALUE_PERSISTENCE_NODE))
                 valPersistenceSettings = new ValuePersistenceSettings(el);
         }

         if (keyPersistenceSettings == null) {
             throw new IllegalArgumentException("Incorrect Cassandra persistence settings specification, " +
                 "there are no key persistence settings specified");
         }

         if (valPersistenceSettings == null) {
             throw new IllegalArgumentException("Incorrect Cassandra persistence settings specification, " +
                 "there are no value persistence settings specified");
         }

         List<PojoKeyField> keyFields = keyPersistenceSettings.getFields();
         List<PojoValueField> valFields = valPersistenceSettings.getFields();

         if (PersistenceStrategy.POJO == keyPersistenceSettings.getStrategy() &&
             (keyFields == null || keyFields.isEmpty())) {
             throw new IllegalArgumentException("Incorrect Cassandra persistence settings specification, " +
                 "there are no key fields found");
         }

         if (PersistenceStrategy.POJO == valPersistenceSettings.getStrategy() &&
             (valFields == null || valFields.isEmpty())) {
             throw new IllegalArgumentException("Incorrect Cassandra persistence settings specification, " +
                 "there are no value fields found");
         }

         // Validating aliases compatibility - fields having different names, but mapped to the same Cassandra table column.
         if (valFields != null && !valFields.isEmpty()) {
             String keyColumn = keyPersistenceSettings.getColumn();
             Class keyClass = keyPersistenceSettings.getJavaClass();

             if (keyColumn != null && !keyColumn.isEmpty()) {
                 for (PojoField valField : valFields) {
                     if (keyColumn.equals(valField.getColumn()) &&
                             !CassandraHelper.isCassandraCompatibleTypes(keyClass, valField.getJavaClass())) {
                         throw new IllegalArgumentException("Value field '" + valField.getName() + "' shares the same " +
                                 "Cassandra table column '" + keyColumn + "' with key, but their Java classes are " +
                                 "different. Fields sharing the same column should have the same Java class as their " +
                                 "type or should be mapped to the same Cassandra primitive type.");
                     }
                 }
             }

             if (keyFields != null && !keyFields.isEmpty()) {
                 for (PojoField keyField : keyFields) {
                     for (PojoField valField : valFields) {
                         if (keyField.getColumn().equals(valField.getColumn()) &&
                                 !CassandraHelper.isCassandraCompatibleTypes(keyField.getJavaClass(), valField.getJavaClass())) {
                             throw new IllegalArgumentException("Value field '" + valField.getName() + "' shares the same " +
                                     "Cassandra table column '" + keyColumn + "' with key field '" + keyField.getName() + "', " +
                                     "but their Java classes are different. Fields sharing the same column should have " +
                                     "the same Java class as their type or should be mapped to the same Cassandra " +
                                     "primitive type.");
                         }
                     }
                 }
             }
         }

         tableColumns = new LinkedList<>();

         for (String column : keyPersistenceSettings.getTableColumns()) {
             if (!tableColumns.contains(column))
                 tableColumns.add(column);
         }

         for (String column : valPersistenceSettings.getTableColumns()) {
             if (!tableColumns.contains(column))
                 tableColumns.add(column);
         }

         tableColumns = Collections.unmodifiableList(tableColumns);
     }
 }
	/*
	* 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.cassandra.persistence;

	import java.io.BufferedReader;
	import java.io.File;
	import java.io.FileInputStream;
	import java.io.IOException;
	import java.io.InputStream;
	import java.io.InputStreamReader;
	import java.io.Serializable;
	import java.io.StringReader;
	import java.util.Collections;
	import java.util.HashSet;
	import java.util.LinkedList;
	import java.util.List;
	import java.util.Set;
	import javax.xml.parsers.DocumentBuilder;
	import javax.xml.parsers.DocumentBuilderFactory;
	import org.apache.ignite.IgniteException;
	import org.apache.ignite.cache.store.cassandra.common.CassandraHelper;
	import org.apache.ignite.cache.store.cassandra.common.SystemHelper;
	import org.apache.ignite.internal.util.typedef.internal.U;
	import org.springframework.core.io.Resource;
	import org.w3c.dom.Document;
	import org.w3c.dom.Element;
	import org.w3c.dom.Node;
	import org.w3c.dom.NodeList;
	import org.xml.sax.InputSource;

	/**
	* Stores persistence settings for Ignite cache key and value
	*/
	public class KeyValuePersistenceSettings implements Serializable {
	/**
	* Default Cassandra keyspace options which should be used to create new keyspace.
	* <ul>
	* <li> <b>SimpleStrategy</b> for replication work well for single data center Cassandra cluster.<br/>
	* If your Cassandra cluster deployed across multiple data centers it's better to use <b>NetworkTopologyStrategy</b>.
	* </li>
	* <li> Three replicas will be created for each data block. </li>
	* <li> Setting DURABLE_WRITES to true specifies that all data should be written to commit log. </li>
	* </ul>
	*/
	private static final String DFLT_KEYSPACE_OPTIONS = "replication = {'class' : 'SimpleStrategy', " +
	"'replication_factor' : 3} and durable_writes = true";

	/** Xml attribute specifying Cassandra keyspace to use. */
	private static final String KEYSPACE_ATTR = "keyspace";

	/** Xml attribute specifying Cassandra table to use. */
	private static final String TABLE_ATTR = "table";

	/** Xml attribute specifying ttl (time to leave) for rows inserted in Cassandra. */
	private static final String TTL_ATTR = "ttl";

	/** Root xml element containing persistence settings specification. */
	private static final String PERSISTENCE_NODE = "persistence";

	/** Xml element specifying Cassandra keyspace options. */
	private static final String KEYSPACE_OPTIONS_NODE = "keyspaceOptions";

	/** Xml element specifying Cassandra table options. */
	private static final String TABLE_OPTIONS_NODE = "tableOptions";

	/** Xml element specifying Ignite cache key persistence settings. */
	private static final String KEY_PERSISTENCE_NODE = "keyPersistence";

	/** Xml element specifying Ignite cache value persistence settings. */
	private static final String VALUE_PERSISTENCE_NODE = "valuePersistence";

	/** TTL (time to leave) for rows inserted into Cassandra table {@link <a href="https://docs.datastax.com/en/cql/3.1/cql/cql_using/use_expire_c.html">Expiring data</a>}. */
	private Integer ttl;

	/** Cassandra keyspace (analog of tablespace in relational databases). */
	private String keyspace;

	/** Cassandra table. */
	private String tbl;

	/** Cassandra table creation options {@link <a href="https://docs.datastax.com/en/cql/3.0/cql/cql_reference/create_table_r.html">CREATE TABLE</a>}. */
	private String tblOptions;

	/** Cassandra keyspace creation options {@link <a href="https://docs.datastax.com/en/cql/3.0/cql/cql_reference/create_keyspace_r.html">CREATE KEYSPACE</a>}. */
	private String keyspaceOptions = DFLT_KEYSPACE_OPTIONS;

	/** Persistence settings for Ignite cache keys. */
	private KeyPersistenceSettings keyPersistenceSettings;

	/** Persistence settings for Ignite cache values. */
	private ValuePersistenceSettings valPersistenceSettings;

	/** List of Cassandra table columns */
	private List<String> tableColumns;

	/**
	* Constructs Ignite cache key/value persistence settings.
	*
	* @param settings string containing xml with persistence settings for Ignite cache key/value
	*/
	public KeyValuePersistenceSettings(String settings) {
	init(settings);
	}

	/**
	* Constructs Ignite cache key/value persistence settings.
	*
	* @param settingsFile xml file with persistence settings for Ignite cache key/value
	*/
	public KeyValuePersistenceSettings(File settingsFile) {
	InputStream in;

	try {
	in = new FileInputStream(settingsFile);
	}
	catch (IOException e) {
	throw new IgniteException("Failed to get input stream for Cassandra persistence settings file: " +
	settingsFile.getAbsolutePath(), e);
	}

	init(loadSettings(in));
	}

	/**
	* Constructs Ignite cache key/value persistence settings.
	*
	* @param settingsRsrc resource containing xml with persistence settings for Ignite cache key/value
	*/
	public KeyValuePersistenceSettings(Resource settingsRsrc) {
	InputStream in;

	try {
	in = settingsRsrc.getInputStream();
	}
	catch (IOException e) {
	throw new IgniteException("Failed to get input stream for Cassandra persistence settings resource: " + settingsRsrc, e);
	}

	init(loadSettings(in));
	}

	/**
	* Returns ttl to use for while inserting new rows into Cassandra table.
	*
	* @return ttl
	*/
	public Integer getTTL() {
	return ttl;
	}

	/**
	* Returns Cassandra keyspace to use.
	*
	* @return keyspace.
	*/
	public String getKeyspace() {
	return keyspace;
	}

	/**
	* Returns Cassandra table to use.
	*
	* @return table.
	*/
	public String getTable() {
	return tbl;
	}

	/**
	* Returns persistence settings for Ignite cache keys.
	*
	* @return keys persistence settings.
	*/
	public KeyPersistenceSettings getKeyPersistenceSettings() {
	return keyPersistenceSettings;
	}

	/**
	* Returns persistence settings for Ignite cache values.
	*
	* @return values persistence settings.
	*/
	public ValuePersistenceSettings getValuePersistenceSettings() {
	return valPersistenceSettings;
	}

	/**
	* Returns list of POJO fields to be mapped to Cassandra table columns.
	*
	* @return POJO fields list.
	*/
	public List<PojoField> getFields() {
	List<PojoField> fields = new LinkedList<>();

	for (PojoField field : keyPersistenceSettings.getFields())
	fields.add(field);

	for (PojoField field : valPersistenceSettings.getFields())
	fields.add(field);

	return fields;
	}

	/**
	* Returns list of Ignite cache key POJO fields to be mapped to Cassandra table columns.
	*
	* @return POJO fields list.
	*/
	public List<PojoKeyField> getKeyFields() {
	return keyPersistenceSettings.getFields();
	}

	/**
	* Returns list of Ignite cache value POJO fields to be mapped to Cassandra table columns.
	*
	* @return POJO fields list.
	*/
	public List<PojoValueField> getValueFields() {
	return valPersistenceSettings.getFields();
	}

	/**
	* Returns DDL statement to create Cassandra keyspace.
	*
	* @return Keyspace DDL statement.
	*/
	public String getKeyspaceDDLStatement() {
	StringBuilder builder = new StringBuilder();
	builder.append("create keyspace if not exists \"").append(keyspace).append("\"");

	if (keyspaceOptions != null) {
	if (!keyspaceOptions.trim().toLowerCase().startsWith("with"))
	builder.append("\nwith");

	builder.append(" ").append(keyspaceOptions);
	}

	String statement = builder.toString().trim().replaceAll(" +", " ");

	return statement.endsWith(";") ? statement : statement + ";";
	}

	/**
	* Returns column names for Cassandra table.
	*
	* @return Column names.
	*/
	public List<String> getTableColumns() {
	return tableColumns;
	}

	/**
	* Returns DDL statement to create Cassandra table.
	*
	* @param table Table name.
	* @return Table DDL statement.
	*/
	public String getTableDDLStatement(String table) {
	if (table == null \|\| table.trim().isEmpty())
	throw new IllegalArgumentException("Table name should be specified");

	String keyColumnsDDL = keyPersistenceSettings.getTableColumnsDDL();
	String valColumnsDDL = valPersistenceSettings.getTableColumnsDDL(new HashSet<>(keyPersistenceSettings.getTableColumns()));

	String colsDDL = keyColumnsDDL;

	if (valColumnsDDL != null && !valColumnsDDL.trim().isEmpty())
	colsDDL += ",\n" + valColumnsDDL;

	String primaryKeyDDL = keyPersistenceSettings.getPrimaryKeyDDL();

	String clusteringDDL = keyPersistenceSettings.getClusteringDDL();

	String optionsDDL = tblOptions != null && !tblOptions.trim().isEmpty() ? tblOptions.trim() : "";

	if (clusteringDDL != null && !clusteringDDL.isEmpty())
	optionsDDL = optionsDDL.isEmpty() ? clusteringDDL : optionsDDL + " and " + clusteringDDL;

	if (!optionsDDL.trim().isEmpty())
	optionsDDL = optionsDDL.trim().toLowerCase().startsWith("with") ? optionsDDL.trim() : "with " + optionsDDL.trim();

	StringBuilder builder = new StringBuilder();

	builder.append("create table if not exists \"").append(keyspace).append("\".\"").append(table).append("\"");
	builder.append("\n(\n").append(colsDDL).append(",\n").append(primaryKeyDDL).append("\n)");

	if (!optionsDDL.isEmpty())
	builder.append(" \n").append(optionsDDL);

	String tblDDL = builder.toString().trim().replaceAll(" +", " ");

	return tblDDL.endsWith(";") ? tblDDL : tblDDL + ";";
	}

	/**
	* Returns DDL statements to create Cassandra table secondary indexes.
	*
	* @param table Table name.
	* @return DDL statements to create secondary indexes.
	*/
	public List<String> getIndexDDLStatements(String table) {
	List<String> idxDDLs = new LinkedList<>();

	Set<String> keyCols = new HashSet<>(keyPersistenceSettings.getTableColumns());

	List<PojoValueField> fields = valPersistenceSettings.getFields();

	for (PojoField field : fields) {
	if (!keyCols.contains(field.getColumn()) && ((PojoValueField)field).isIndexed())
	idxDDLs.add(((PojoValueField)field).getIndexDDL(keyspace, table));
	}

	return idxDDLs;
	}

	/**
	* Loads Ignite cache persistence settings from resource.
	*
	* @param in Input stream.
	* @return String containing xml with Ignite cache persistence settings.
	*/
	private String loadSettings(InputStream in) {
	StringBuilder settings = new StringBuilder();
	BufferedReader reader = null;

	try {
	reader = new BufferedReader(new InputStreamReader(in));

	String line = reader.readLine();

	while (line != null) {
	if (settings.length() != 0)
	settings.append(SystemHelper.LINE_SEPARATOR);

	settings.append(line);

	line = reader.readLine();
	}
	}
	catch (Throwable e) {
	throw new IgniteException("Failed to read input stream for Cassandra persistence settings", e);
	}
	finally {
	U.closeQuiet(reader);
	U.closeQuiet(in);
	}

	return settings.toString();
	}

	/**
	* @param elem Element with data.
	* @param attr Attribute name.
	* @return Numeric value for specified attribute.
	*/
	private int extractIntAttribute(Element elem, String attr) {
	String val = elem.getAttribute(attr).trim();

	try {
	return Integer.parseInt(val);
	}
	catch (NumberFormatException ignored) {
	throw new IllegalArgumentException("Incorrect value '" + val + "' specified for '" + attr + "' attribute");
	}
	}

	/**
	* Initializes persistence settings from XML string.
	*
	* @param settings XML string containing Ignite cache persistence settings configuration.
	*/
	@SuppressWarnings("IfCanBeSwitch")
	private void init(String settings) {
	Document doc;

	try {
	DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance();
	DocumentBuilder builder = factory.newDocumentBuilder();
	doc = builder.parse(new InputSource(new StringReader(settings)));
	}
	catch (Throwable e) {
	throw new IllegalArgumentException("Failed to parse persistence settings:" +
	SystemHelper.LINE_SEPARATOR + settings, e);
	}

	Element root = doc.getDocumentElement();

	if (!PERSISTENCE_NODE.equals(root.getNodeName())) {
	throw new IllegalArgumentException("Incorrect persistence settings specified. " +
	"Root XML element should be 'persistence'");
	}

	if (!root.hasAttribute(KEYSPACE_ATTR)) {
	throw new IllegalArgumentException("Incorrect persistence settings '" + KEYSPACE_ATTR +
	"' attribute should be specified");
	}

	keyspace = root.getAttribute(KEYSPACE_ATTR).trim();
	tbl = root.hasAttribute(TABLE_ATTR) ? root.getAttribute(TABLE_ATTR).trim() : null;

	if (root.hasAttribute(TTL_ATTR))
	ttl = extractIntAttribute(root, TTL_ATTR);

	if (!root.hasChildNodes()) {
	throw new IllegalArgumentException("Incorrect Cassandra persistence settings specification, " +
	"there are no key and value persistence settings specified");
	}

	NodeList children = root.getChildNodes();
	int cnt = children.getLength();

	for (int i = 0; i < cnt; i++) {
	Node node = children.item(i);

	if (node.getNodeType() != Node.ELEMENT_NODE)
	continue;

	Element el = (Element)node;
	String nodeName = el.getNodeName();

	if (nodeName.equals(TABLE_OPTIONS_NODE)) {
	tblOptions = el.getTextContent();
	tblOptions = tblOptions.replace("\n", " ").replace("\r", "").replace("\t", " ");
	}
	else if (nodeName.equals(KEYSPACE_OPTIONS_NODE)) {
	keyspaceOptions = el.getTextContent();
	keyspaceOptions = keyspaceOptions.replace("\n", " ").replace("\r", "").replace("\t", " ");
	}
	else if (nodeName.equals(KEY_PERSISTENCE_NODE))
	keyPersistenceSettings = new KeyPersistenceSettings(el);
	else if (nodeName.equals(VALUE_PERSISTENCE_NODE))
	valPersistenceSettings = new ValuePersistenceSettings(el);
	}

	if (keyPersistenceSettings == null) {
	throw new IllegalArgumentException("Incorrect Cassandra persistence settings specification, " +
	"there are no key persistence settings specified");
	}

	if (valPersistenceSettings == null) {
	throw new IllegalArgumentException("Incorrect Cassandra persistence settings specification, " +
	"there are no value persistence settings specified");
	}

	List<PojoKeyField> keyFields = keyPersistenceSettings.getFields();
	List<PojoValueField> valFields = valPersistenceSettings.getFields();

	if (PersistenceStrategy.POJO == keyPersistenceSettings.getStrategy() &&
	(keyFields == null \|\| keyFields.isEmpty())) {
	throw new IllegalArgumentException("Incorrect Cassandra persistence settings specification, " +
	"there are no key fields found");
	}

	if (PersistenceStrategy.POJO == valPersistenceSettings.getStrategy() &&
	(valFields == null \|\| valFields.isEmpty())) {
	throw new IllegalArgumentException("Incorrect Cassandra persistence settings specification, " +
	"there are no value fields found");
	}

	// Validating aliases compatibility - fields having different names, but mapped to the same Cassandra table column.
	if (valFields != null && !valFields.isEmpty()) {
	String keyColumn = keyPersistenceSettings.getColumn();
	Class keyClass = keyPersistenceSettings.getJavaClass();

	if (keyColumn != null && !keyColumn.isEmpty()) {
	for (PojoField valField : valFields) {
	if (keyColumn.equals(valField.getColumn()) &&
	!CassandraHelper.isCassandraCompatibleTypes(keyClass, valField.getJavaClass())) {
	throw new IllegalArgumentException("Value field '" + valField.getName() + "' shares the same " +
	"Cassandra table column '" + keyColumn + "' with key, but their Java classes are " +
	"different. Fields sharing the same column should have the same Java class as their " +
	"type or should be mapped to the same Cassandra primitive type.");
	}
	}
	}

	if (keyFields != null && !keyFields.isEmpty()) {
	for (PojoField keyField : keyFields) {
	for (PojoField valField : valFields) {
	if (keyField.getColumn().equals(valField.getColumn()) &&
	!CassandraHelper.isCassandraCompatibleTypes(keyField.getJavaClass(), valField.getJavaClass())) {
	throw new IllegalArgumentException("Value field '" + valField.getName() + "' shares the same " +
	"Cassandra table column '" + keyColumn + "' with key field '" + keyField.getName() + "', " +
	"but their Java classes are different. Fields sharing the same column should have " +
	"the same Java class as their type or should be mapped to the same Cassandra " +
	"primitive type.");
	}
	}
	}
	}
	}

	tableColumns = new LinkedList<>();

	for (String column : keyPersistenceSettings.getTableColumns()) {
	if (!tableColumns.contains(column))
	tableColumns.add(column);
	}

	for (String column : valPersistenceSettings.getTableColumns()) {
	if (!tableColumns.contains(column))
	tableColumns.add(column);
	}

	tableColumns = Collections.unmodifiableList(tableColumns);
	}
	}