gora-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From lewi...@apache.org
Subject [04/13] gora git commit: Rename gora-cassandra-cql module into gora-cassandra module
Date Wed, 23 Aug 2017 22:16:35 GMT
http://git-wip-us.apache.org/repos/asf/gora/blob/89683c74/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/CassandraQueryFactory.java
----------------------------------------------------------------------
diff --git a/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/CassandraQueryFactory.java b/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/CassandraQueryFactory.java
new file mode 100644
index 0000000..928370c
--- /dev/null
+++ b/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/CassandraQueryFactory.java
@@ -0,0 +1,836 @@
+/*
+ *  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.gora.cassandra.serializers;
+
+import com.datastax.driver.core.querybuilder.BuiltStatement;
+import com.datastax.driver.core.querybuilder.Delete;
+import com.datastax.driver.core.querybuilder.QueryBuilder;
+import com.datastax.driver.core.querybuilder.Select;
+import com.datastax.driver.core.querybuilder.Update;
+import com.datastax.driver.mapping.annotations.UDT;
+import org.apache.avro.Schema;
+import org.apache.gora.cassandra.bean.CassandraKey;
+import org.apache.gora.cassandra.bean.ClusterKeyField;
+import org.apache.gora.cassandra.bean.Field;
+import org.apache.gora.cassandra.bean.KeySpace;
+import org.apache.gora.cassandra.bean.PartitionKeyField;
+import org.apache.gora.cassandra.query.CassandraQuery;
+import org.apache.gora.cassandra.store.CassandraMapping;
+import org.apache.gora.cassandra.store.CassandraStore;
+import org.apache.gora.query.Query;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * This class is used create Cassandra Queries.
+ */
+class CassandraQueryFactory {
+
+  private static final Logger LOG = LoggerFactory.getLogger(CassandraQueryFactory.class);
+
+  /**
+   * This method returns the CQL query to create key space.
+   * refer : http://docs.datastax.com/en/cql/3.1/cql/cql_reference/create_keyspace_r.html
+   *
+   * @param mapping Cassandra Mapping {@link CassandraMapping}
+   * @return CQL Query
+   */
+  static String getCreateKeySpaceQuery(CassandraMapping mapping) {
+    KeySpace keySpace = mapping.getKeySpace();
+    StringBuilder stringBuffer = new StringBuilder();
+    stringBuffer.append("CREATE KEYSPACE IF NOT EXISTS ").append(keySpace.getName()).append(" WITH REPLICATION = { 'class' : ");
+    KeySpace.PlacementStrategy placementStrategy = keySpace.getPlacementStrategy();
+    stringBuffer.append("'").append(placementStrategy).append("'").append(", ").append("'");
+    switch (placementStrategy) {
+      case SimpleStrategy:
+        stringBuffer.append("replication_factor").append("'").append(" : ").append(keySpace.getReplicationFactor()).append(" }");
+        break;
+      case NetworkTopologyStrategy:
+        boolean isCommaNeeded = false;
+        for (Map.Entry<String, Integer> entry : keySpace.getDataCenters().entrySet()) {
+          if (isCommaNeeded) {
+            stringBuffer.append(", '");
+          }
+          stringBuffer.append(entry.getKey()).append("'").append(" : ").append(entry.getValue());
+          isCommaNeeded = true;
+        }
+        stringBuffer.append(" }");
+        break;
+    }
+    if (keySpace.isDurableWritesEnabled()) {
+      stringBuffer.append(" AND DURABLE_WRITES = ").append(keySpace.isDurableWritesEnabled());
+    }
+    return stringBuffer.toString();
+  }
+
+  /**
+   * This method returns the CQL query to table.
+   * refer : http://docs.datastax.com/en/cql/3.1/cql/cql_reference/create_table_r.html
+   * <p>
+   * Trick : To have a consistency of the order of the columns, first we append partition keys, second cluster keys and finally other columns.
+   * It's very much needed to follow the same order in other CRUD operations as well.
+   *
+   * @param mapping Cassandra mapping {@link CassandraMapping}
+   * @return CQL Query
+   */
+  static String getCreateTableQuery(CassandraMapping mapping) {
+    StringBuilder stringBuffer = new StringBuilder();
+    stringBuffer.append("CREATE TABLE IF NOT EXISTS ").append(mapping.getKeySpace().getName()).append(".").append(mapping.getCoreName()).append(" (");
+    CassandraKey cassandraKey = mapping.getCassandraKey();
+    // appending Cassandra Persistent columns into db schema
+    processFieldsForCreateTableQuery(mapping.getFieldList(), false, stringBuffer);
+
+    if (cassandraKey != null) {
+      processFieldsForCreateTableQuery(cassandraKey.getFieldList(), true, stringBuffer);
+      List<PartitionKeyField> partitionKeys = cassandraKey.getPartitionKeyFields();
+      if (partitionKeys != null) {
+        stringBuffer.append(", PRIMARY KEY (");
+        boolean isCommaNeededToApply = false;
+        for (PartitionKeyField keyField : partitionKeys) {
+          if (isCommaNeededToApply) {
+            stringBuffer.append(",");
+          }
+          if (keyField.isComposite()) {
+            stringBuffer.append("(");
+            boolean isCommaNeededHere = false;
+            for (Field field : keyField.getFields()) {
+              if (isCommaNeededHere) {
+                stringBuffer.append(", ");
+              }
+              stringBuffer.append(field.getColumnName());
+              isCommaNeededHere = true;
+            }
+            stringBuffer.append(")");
+          } else {
+            stringBuffer.append(keyField.getColumnName());
+          }
+          isCommaNeededToApply = true;
+        }
+        stringBuffer.append(")");
+      }
+    }
+
+    stringBuffer.append(")");
+    boolean isWithNeeded = true;
+    if (Boolean.parseBoolean(mapping.getProperty("compactStorage"))) {
+      stringBuffer.append(" WITH COMPACT STORAGE ");
+      isWithNeeded = false;
+    }
+
+    String id = mapping.getProperty("id");
+    if (id != null) {
+      if (isWithNeeded) {
+        stringBuffer.append(" WITH ");
+      } else {
+        stringBuffer.append(" AND ");
+      }
+      stringBuffer.append("ID = '").append(id).append("'");
+      isWithNeeded = false;
+    }
+    if (cassandraKey != null) {
+      List<ClusterKeyField> clusterKeyFields = cassandraKey.getClusterKeyFields();
+      if (clusterKeyFields != null) {
+        if (isWithNeeded) {
+          stringBuffer.append(" WITH ");
+        } else {
+          stringBuffer.append(" AND ");
+        }
+        stringBuffer.append(" CLUSTERING ORDER BY (");
+        boolean isCommaNeededToApply = false;
+        for (ClusterKeyField keyField : clusterKeyFields) {
+          if (isCommaNeededToApply) {
+            stringBuffer.append(", ");
+          }
+          stringBuffer.append(keyField.getColumnName()).append(" ");
+          if (keyField.getOrder() != null) {
+            stringBuffer.append(keyField.getOrder());
+          }
+          isCommaNeededToApply = true;
+        }
+        stringBuffer.append(")");
+      }
+    }
+    return stringBuffer.toString();
+  }
+
+  private static void processFieldsForCreateTableQuery(List<Field> fields, boolean isCommaNeeded, StringBuilder stringBuilder) {
+    for (Field field : fields) {
+      if (isCommaNeeded) {
+        stringBuilder.append(", ");
+      }
+      stringBuilder.append(field.getColumnName()).append(" ").append(field.getType());
+      boolean isStaticColumn = Boolean.parseBoolean(field.getProperty("static"));
+      boolean isPrimaryKey = Boolean.parseBoolean(field.getProperty("primarykey"));
+      if (isStaticColumn) {
+        stringBuilder.append(" STATIC");
+      }
+      if (isPrimaryKey) {
+        stringBuilder.append("  PRIMARY KEY ");
+      }
+      isCommaNeeded = true;
+    }
+  }
+
+  /**
+   * This method returns the CQL query to drop table.
+   * refer : http://docs.datastax.com/en/cql/3.1/cql/cql_reference/drop_table_r.html
+   *
+   * @param mapping Cassandra Mapping {@link CassandraMapping}
+   * @return CQL query
+   */
+  static String getDropTableQuery(CassandraMapping mapping) {
+    return "DROP TABLE IF EXISTS " + mapping.getKeySpace().getName() + "." + mapping.getCoreName();
+  }
+
+  /**
+   * This method returns the CQL query to drop key space.
+   * refer : http://docs.datastax.com/en/cql/3.1/cql/cql_reference/drop_keyspace_r.html
+   *
+   * @param mapping Cassandra Mapping {@link CassandraMapping}
+   * @return CQL query
+   */
+  static String getDropKeySpaceQuery(CassandraMapping mapping) {
+    return "DROP KEYSPACE IF EXISTS " + mapping.getKeySpace().getName();
+  }
+
+  /**
+   * This method returns the CQL query to truncate (removes all the data) in the table.
+   * refer : http://docs.datastax.com/en/cql/3.1/cql/cql_reference/truncate_r.html
+   *
+   * @param mapping Cassandra Mapping {@link CassandraMapping}
+   * @return CQL query
+   */
+  static String getTruncateTableQuery(CassandraMapping mapping) {
+    return QueryBuilder.truncate(mapping.getKeySpace().getName(), mapping.getCoreName()).getQueryString();
+  }
+
+  /**
+   * This method return the CQL query to insert data in to the table.
+   * refer : http://docs.datastax.com/en/cql/3.1/cql/cql_reference/insert_r.html
+   *
+   * @param mapping Cassandra Mapping {@link CassandraMapping}
+   * @param fields  available fields
+   * @return CQL Query
+   */
+  static String getInsertDataQuery(CassandraMapping mapping, List<String> fields) {
+    String[] columnNames = getColumnNames(mapping, fields);
+    String[] objects = new String[fields.size()];
+    Arrays.fill(objects, "?");
+    return QueryBuilder.insertInto(mapping.getKeySpace().getName(), mapping.getCoreName()).values(columnNames, objects).getQueryString();
+  }
+
+  /**
+   * This method return the CQL query to delete a persistent in the table.
+   * refer : http://docs.datastax.com/en/cql/3.3/cql/cql_reference/cqlDelete.html
+   *
+   * @param mapping Cassandra Mapping {@link CassandraMapping}
+   * @param fields  filed list to be deleted
+   * @return CQL Query
+   */
+  static String getDeleteDataQuery(CassandraMapping mapping, List<String> fields) {
+    String[] columnNames = getColumnNames(mapping, fields);
+    String[] objects = new String[fields.size()];
+    Arrays.fill(objects, "?");
+    Delete delete = QueryBuilder.delete().from(mapping.getKeySpace().getName(), mapping.getCoreName());
+    return processKeys(columnNames, delete);
+  }
+
+  private static String processKeys(String[] columnNames, BuiltStatement delete) {
+    BuiltStatement query = null;
+    boolean isWhereNeeded = true;
+    for (String columnName : columnNames) {
+      if (isWhereNeeded) {
+        if (delete instanceof Delete) {
+          query = ((Delete) delete).where(QueryBuilder.eq(columnName, "?"));
+        } else {
+          query = ((Select) delete).where(QueryBuilder.eq(columnName, "?"));
+        }
+        isWhereNeeded = false;
+      } else {
+        if (delete instanceof Delete) {
+          query = ((Delete.Where) query).and(QueryBuilder.eq(columnName, "?"));
+        } else {
+          query = ((Select.Where) query).and(QueryBuilder.eq(columnName, "?"));
+        }
+      }
+    }
+    return query != null ? query.getQueryString() : null;
+  }
+
+  /**
+   * This method returns the CQL Select query to retrieve data from the table.
+   * refer: http://docs.datastax.com/en/cql/3.3/cql/cql_reference/cqlSelect.html
+   *
+   * @param mapping   Cassandra Mapping {@link CassandraMapping}
+   * @param keyFields key fields
+   * @return CQL Query
+   */
+  static String getSelectObjectQuery(CassandraMapping mapping, List<String> keyFields) {
+    Select select = QueryBuilder.select().from(mapping.getKeySpace().getName(), mapping.getCoreName());
+    if (Boolean.parseBoolean(mapping.getProperty("allowFiltering"))) {
+      select.allowFiltering();
+    }
+    String[] columnNames = getColumnNames(mapping, keyFields);
+    return processKeys(columnNames, select);
+  }
+
+  /**
+   * This method returns CQL Select query to retrieve data from the table with given fields.
+   * This method is used for Avro Serialization
+   * refer: http://docs.datastax.com/en/cql/3.3/cql/cql_reference/cqlSelect.html
+   *
+   * @param mapping   Cassandra Mapping {@link CassandraMapping}
+   * @param fields    Given fields to retrieve
+   * @param keyFields key fields
+   * @return CQL Query
+   */
+  static String getSelectObjectWithFieldsQuery(CassandraMapping mapping, String[] fields, List<String> keyFields) {
+    Select select = QueryBuilder.select(getColumnNames(mapping, Arrays.asList(fields))).from(mapping.getKeySpace().getName(), mapping.getCoreName());
+    if (Boolean.parseBoolean(mapping.getProperty("allowFiltering"))) {
+      select.allowFiltering();
+    }
+    String[] columnNames = getColumnNames(mapping, keyFields);
+    return processKeys(columnNames, select);
+  }
+
+  /**
+   * This method returns CQL Select query to retrieve data from the table with given fields.
+   * This method is used for Native Serialization
+   * refer: http://docs.datastax.com/en/cql/3.3/cql/cql_reference/cqlSelect.html
+   *
+   * @param mapping Cassandra Mapping {@link CassandraMapping}
+   * @param fields  Given fields to retrieve
+   * @return CQL Query
+   */
+  static String getSelectObjectWithFieldsQuery(CassandraMapping mapping, String[] fields) {
+    String cqlQuery = null;
+    String[] columnNames = getColumnNames(mapping, Arrays.asList(fields));
+    Select select = QueryBuilder.select(columnNames).from(mapping.getKeySpace().getName(), mapping.getCoreName());
+    if (Boolean.parseBoolean(mapping.getProperty("allowFiltering"))) {
+      select.allowFiltering();
+    }
+    CassandraKey cKey = mapping.getCassandraKey();
+    if (cKey != null) {
+      Select.Where query = null;
+      boolean isWhereNeeded = true;
+      for (Field field : cKey.getFieldList()) {
+        if (isWhereNeeded) {
+          query = select.where(QueryBuilder.eq(field.getColumnName(), "?"));
+          isWhereNeeded = false;
+        } else {
+          query = query.and(QueryBuilder.eq(field.getColumnName(), "?"));
+        }
+      }
+      cqlQuery = query != null ? query.getQueryString() : null;
+    } else {
+      for (Field field : mapping.getFieldList()) {
+        boolean isPrimaryKey = Boolean.parseBoolean(field.getProperty("primarykey"));
+        if (isPrimaryKey) {
+          cqlQuery = select.where(QueryBuilder.eq(field.getColumnName(), "?")).getQueryString();
+          break;
+        }
+      }
+    }
+    return cqlQuery;
+  }
+
+
+  /**
+   * This method returns CQL Query for execute method. This CQL contains a Select Query to retrieve data from the table
+   *
+   * @param mapping        Cassandra Mapping {@link CassandraMapping}
+   * @param cassandraQuery Query {@link CassandraQuery}
+   * @param objects        object list
+   * @return CQL Query
+   */
+  static String getExecuteQuery(CassandraMapping mapping, Query cassandraQuery, List<Object> objects, String[] fields) {
+    long limit = cassandraQuery.getLimit();
+    Select select = QueryBuilder.select(getColumnNames(mapping, Arrays.asList(fields))).from(mapping.getKeySpace().getName(), mapping.getCoreName());
+    if (limit > 0) {
+      select = select.limit((int) limit);
+    }
+    if (Boolean.parseBoolean(mapping.getProperty("allowFiltering"))) {
+      select.allowFiltering();
+    }
+    return processQuery(cassandraQuery, select, mapping, objects);
+  }
+
+  private static String processQuery(Query cassandraQuery, BuiltStatement select, CassandraMapping mapping, List<Object> objects) {
+    String primaryKey = null;
+    BuiltStatement query = null;
+    Object startKey = cassandraQuery.getStartKey();
+    Object endKey = cassandraQuery.getEndKey();
+    Object key = cassandraQuery.getKey();
+    boolean isWhereNeeded = true;
+    if (key != null) {
+      if (mapping.getCassandraKey() != null) {
+        ArrayList<String> cassandraKeys = new ArrayList<>();
+        ArrayList<Object> cassandraValues = new ArrayList<>();
+        AvroCassandraUtils.processKeys(mapping, key, cassandraKeys, cassandraValues);
+        String[] columnKeys = getColumnNames(mapping, cassandraKeys);
+        for (int i = 0; i < cassandraKeys.size(); i++) {
+          if (isWhereNeeded) {
+            if (select instanceof Select) {
+              query = ((Select) select).where(QueryBuilder.eq(columnKeys[i], "?"));
+            } else if (select instanceof Delete) {
+              query = ((Delete) select).where(QueryBuilder.eq(columnKeys[i], "?"));
+            } else {
+              query = ((Update.Assignments) select).where(QueryBuilder.eq(columnKeys[i], "?"));
+            }
+            objects.add(cassandraValues.get(i));
+            isWhereNeeded = false;
+          } else {
+            if (select instanceof Select) {
+              query = ((Select.Where) query).and(QueryBuilder.eq(columnKeys[i], "?"));
+            } else if (select instanceof Delete) {
+              query = ((Delete.Where) query).and(QueryBuilder.eq(columnKeys[i], "?"));
+            } else {
+              query = ((Update.Where) query).and(QueryBuilder.eq(columnKeys[i], "?"));
+            }
+            objects.add(cassandraValues.get(i));
+          }
+        }
+      } else {
+        primaryKey = getPKey(mapping.getFieldList());
+        if (select instanceof Select) {
+          query = ((Select) select).where(QueryBuilder.eq(primaryKey, "?"));
+        } else if (select instanceof Delete) {
+          query = ((Delete) select).where(QueryBuilder.eq(primaryKey, "?"));
+        } else {
+          query = ((Update.Assignments) select).where(QueryBuilder.eq(primaryKey, "?"));
+        }
+        objects.add(key);
+      }
+    } else {
+      if (startKey != null) {
+        if (mapping.getCassandraKey() != null) {
+          ArrayList<String> cassandraKeys = new ArrayList<>();
+          ArrayList<Object> cassandraValues = new ArrayList<>();
+          AvroCassandraUtils.processKeys(mapping, startKey, cassandraKeys, cassandraValues);
+          String[] columnKeys = getColumnNames(mapping, cassandraKeys);
+          for (int i = 0; i < cassandraKeys.size(); i++) {
+            if (isWhereNeeded) {
+              if (select instanceof Select) {
+                query = ((Select) select).where(QueryBuilder.gte(columnKeys[i], "?"));
+              } else if (select instanceof Delete) {
+                /*
+                According to the JIRA https://issues.apache.org/jira/browse/CASSANDRA-7651 this has been fixed, but It seems this not fixed yet.
+                 */
+                throw new RuntimeException("Delete by Query is not suppoted for Key Ranges.");
+              } else {
+                query = ((Update.Assignments) select).where(QueryBuilder.gte(columnKeys[i], "?"));
+              }
+              objects.add(cassandraValues.get(i));
+              isWhereNeeded = false;
+            } else {
+              if (select instanceof Select) {
+                query = ((Select.Where) query).and(QueryBuilder.gte(columnKeys[i], "?"));
+              } else if (select instanceof Delete) {
+                       /*
+                According to the JIRA https://issues.apache.org/jira/browse/CASSANDRA-7651 this has been fixed, but It seems this not fixed yet.
+                 */
+                throw new RuntimeException("Delete by Query is not suppoted for Key Ranges.");
+              } else {
+                query = ((Update.Where) query).and(QueryBuilder.gte(columnKeys[i], "?"));
+              }
+              objects.add(cassandraValues.get(i));
+            }
+          }
+        } else {
+          primaryKey = getPKey(mapping.getFieldList());
+          if (select instanceof Select) {
+            query = ((Select) select).where(QueryBuilder.gte(primaryKey, "?"));
+          } else if (select instanceof Delete) {
+                           /*
+                According to the JIRA https://issues.apache.org/jira/browse/CASSANDRA-7651 this has been fixed, but It seems this not fixed yet.
+                 */
+            throw new RuntimeException("Delete by Query is not suppoted for Key Ranges.");
+          } else {
+            query = ((Update.Assignments) select).where(QueryBuilder.gte(primaryKey, "?"));
+          }
+          objects.add(startKey);
+          isWhereNeeded = false;
+        }
+      }
+      if (endKey != null) {
+        if (mapping.getCassandraKey() != null) {
+          ArrayList<String> cassandraKeys = new ArrayList<>();
+          ArrayList<Object> cassandraValues = new ArrayList<>();
+          AvroCassandraUtils.processKeys(mapping, endKey, cassandraKeys, cassandraValues);
+          String[] columnKeys = getColumnNames(mapping, cassandraKeys);
+          for (int i = 0; i < cassandraKeys.size(); i++) {
+            if (isWhereNeeded) {
+              if (select instanceof Select) {
+                query = ((Select) select).where(QueryBuilder.lte(columnKeys[i], "?"));
+              } else if (select instanceof Delete) {
+                               /*
+                According to the JIRA https://issues.apache.org/jira/browse/CASSANDRA-7651 this has been fixed, but It seems this not fixed yet.
+                 */
+                throw new RuntimeException("Delete by Query is not suppoted for Key Ranges.");
+              } else {
+                query = ((Update.Assignments) select).where(QueryBuilder.lte(columnKeys[i], "?"));
+              }
+              objects.add(cassandraValues.get(i));
+              isWhereNeeded = false;
+            } else {
+              if (select instanceof Select) {
+                query = ((Select.Where) query).and(QueryBuilder.lte(columnKeys[i], "?"));
+              } else if (select instanceof Delete) {
+                               /*
+                According to the JIRA https://issues.apache.org/jira/browse/CASSANDRA-7651 this has been fixed, but It seems this not fixed yet.
+                 */
+                throw new RuntimeException("Delete by Query is not suppoted for Key Ranges.");
+              } else {
+                query = ((Update.Where) query).and(QueryBuilder.lte(columnKeys[i], "?"));
+              }
+              objects.add(cassandraValues.get(i));
+            }
+          }
+        } else {
+          primaryKey = primaryKey != null ? primaryKey : getPKey(mapping.getFieldList());
+          if (isWhereNeeded) {
+            if (select instanceof Select) {
+              query = ((Select) select).where(QueryBuilder.lte(primaryKey, "?"));
+            } else if (select instanceof Delete) {
+                             /*
+                According to the JIRA https://issues.apache.org/jira/browse/CASSANDRA-7651 this has been fixed, but It seems this not fixed yet.
+                 */
+              throw new RuntimeException("Delete by Query is not suppoted for Key Ranges.");
+            } else {
+              query = ((Update.Assignments) select).where(QueryBuilder.lte(primaryKey, "?"));
+            }
+          } else {
+            if (select instanceof Select) {
+              query = ((Select.Where) query).and(QueryBuilder.lte(primaryKey, "?"));
+            } else if (select instanceof Delete) {
+              /*
+                According to the JIRA https://issues.apache.org/jira/browse/CASSANDRA-7651 this has been fixed, but It seems this not fixed yet.
+                */
+              throw new RuntimeException("Delete by Query is not suppoted for Key Ranges.");
+            } else {
+              query = ((Update.Where) query).and(QueryBuilder.lte(primaryKey, "?"));
+            }
+          }
+          objects.add(endKey);
+        }
+      }
+    }
+    if (startKey == null && endKey == null && key == null) {
+      return select.getQueryString();
+    }
+    return query != null ? query.getQueryString() : null;
+  }
+
+  private static String[] getColumnNames(CassandraMapping mapping, List<String> fields) {
+    ArrayList<String> columnNames = new ArrayList<>();
+    for (String field : fields) {
+      Field fieldBean = mapping.getFieldFromFieldName(field);
+      CassandraKey cassandraKey = mapping.getCassandraKey();
+      Field keyBean = null;
+      if (cassandraKey != null) {
+        keyBean = cassandraKey.getFieldFromFieldName(field);
+      }
+      if (fieldBean != null) {
+        columnNames.add(fieldBean.getColumnName());
+      } else if (keyBean != null) {
+        columnNames.add(keyBean.getColumnName());
+      } else {
+        LOG.warn("{} field is ignored, couldn't find relevant field in the persistent mapping", field);
+      }
+    }
+    return columnNames.toArray(new String[0]);
+  }
+
+  private static String getPKey(List<Field> fields) {
+    for (Field field : fields) {
+      boolean isPrimaryKey = Boolean.parseBoolean(field.getProperty("primarykey"));
+      if (isPrimaryKey) {
+        return field.getColumnName();
+      }
+    }
+    return null;
+  }
+
+  /**
+   * This method returns CQL Qeury for DeleteByQuery method.
+   * refer: http://docs.datastax.com/en/cql/3.3/cql/cql_reference/cqlDelete.html
+   *
+   * @param mapping        Cassandra Mapping {@link CassandraMapping}
+   * @param cassandraQuery Cassandra Query {@link CassandraQuery}
+   * @param objects        field values
+   * @return CQL Query
+   */
+  static String getDeleteByQuery(CassandraMapping mapping, Query cassandraQuery, List<Object> objects) {
+    String[] columns = null;
+    if (cassandraQuery.getFields() != null) {
+      columns = getColumnNames(mapping, Arrays.asList(cassandraQuery.getFields()));
+    }
+    Delete delete;
+    if (columns != null) {
+      delete = QueryBuilder.delete(columns).from(mapping.getKeySpace().getName(), mapping.getCoreName());
+    } else {
+      delete = QueryBuilder.delete().from(mapping.getKeySpace().getName(), mapping.getCoreName());
+    }
+    return processQuery(cassandraQuery, delete, mapping, objects);
+  }
+
+  /**
+   * This method returns the CQL Query for UpdateByQuery method
+   * refer : http://docs.datastax.com/en/cql/3.3/cql/cql_reference/cqlUpdate.html
+   *
+   * @param mapping        Cassandra mapping {@link CassandraMapping}
+   * @param cassandraQuery Cassandra Query {@link CassandraQuery}
+   * @param objects        field Objects list
+   * @return CQL Query
+   */
+  static String getUpdateByQueryForAvro(CassandraMapping mapping, Query cassandraQuery, List<Object> objects, Schema schema) {
+    Update update = QueryBuilder.update(mapping.getKeySpace().getName(), mapping.getCoreName());
+    Update.Assignments updateAssignments = null;
+    if (cassandraQuery instanceof CassandraQuery) {
+      String[] columnNames = getColumnNames(mapping, Arrays.asList(cassandraQuery.getFields()));
+        for (String column : columnNames) {
+          updateAssignments = update.with(QueryBuilder.set(column, "?"));
+          Field field = mapping.getFieldFromColumnName(column);
+          Object value = ((CassandraQuery) cassandraQuery).getUpdateFieldValue(field.getFieldName());
+          try {
+            Schema schemaField = schema.getField(field.getFieldName()).schema();
+            objects.add(AvroCassandraUtils.getFieldValueFromAvroBean(schemaField, schemaField.getType(), value, field));
+          } catch (NullPointerException e) {
+            throw new RuntimeException(field + " field couldn't find in the class " + mapping.getPersistentClass() + ".");
+          }
+        }
+    } else {
+      throw new RuntimeException("Please use Cassandra Query object to invoke, UpdateByQuery method.");
+    }
+    return processQuery(cassandraQuery, updateAssignments, mapping, objects);
+  }
+
+
+  /**
+   * This method returns the CQL Query for UpdateByQuery method
+   * refer : http://docs.datastax.com/en/cql/3.3/cql/cql_reference/cqlUpdate.html
+   *
+   * @param mapping        Cassandra mapping {@link CassandraMapping}
+   * @param cassandraQuery Cassandra Query {@link CassandraQuery}
+   * @param objects        field Objects list
+   * @return CQL Query
+   */
+  static String getUpdateByQueryForNative(CassandraMapping mapping, Query cassandraQuery, List<Object> objects) {
+    Update update = QueryBuilder.update(mapping.getKeySpace().getName(), mapping.getCoreName());
+    Update.Assignments updateAssignments = null;
+    if (cassandraQuery instanceof CassandraQuery) {
+      String[] columnNames = getColumnNames(mapping, Arrays.asList(cassandraQuery.getFields()));
+        for (String column : columnNames) {
+          updateAssignments = update.with(QueryBuilder.set(column, "?"));
+          objects.add(((CassandraQuery) cassandraQuery).getUpdateFieldValue(mapping.getFieldFromColumnName(column).getFieldName()));
+        }
+    } else {
+      throw new RuntimeException("Please use Cassandra Query object to invoke, UpdateByQuery method.");
+    }
+    return processQuery(cassandraQuery, updateAssignments, mapping, objects);
+  }
+
+
+  private static void populateFieldsToQuery(Schema schema, StringBuilder builder) throws Exception {
+    switch (schema.getType()) {
+      case INT:
+        builder.append("int");
+        break;
+      case MAP:
+        builder.append("map<text,");
+        populateFieldsToQuery(schema.getValueType(), builder);
+        builder.append(">");
+        break;
+      case ARRAY:
+        builder.append("list<");
+        populateFieldsToQuery(schema.getElementType(), builder);
+        builder.append(">");
+        break;
+      case LONG:
+        builder.append("bigint");
+        break;
+      case FLOAT:
+        builder.append("float");
+        break;
+      case DOUBLE:
+        builder.append("double");
+        break;
+      case BOOLEAN:
+        builder.append("boolean");
+        break;
+      case BYTES:
+        builder.append("blob");
+        break;
+      case RECORD:
+        builder.append("frozen<").append(schema.getName()).append(">");
+        break;
+      case STRING:
+      case FIXED:
+      case ENUM:
+        builder.append("text");
+        break;
+      case UNION:
+        for (Schema unionElementSchema : schema.getTypes()) {
+          if (unionElementSchema.getType().equals(Schema.Type.RECORD)) {
+            String recordName = unionElementSchema.getName();
+            if (!builder.toString().contains(recordName)) {
+              builder.append("frozen<").append(recordName).append(">");
+            } else {
+              LOG.warn("Same Field Type can't be mapped recursively. This is not supported with Cassandra UDT types, Please use byte dataType for recursive mapping.");
+              throw new Exception("Same Field Type has mapped recursively");
+            }
+            break;
+          } else if (!unionElementSchema.getType().equals(Schema.Type.NULL)) {
+            populateFieldsToQuery(unionElementSchema, builder);
+            break;
+          }
+        }
+        break;
+    }
+  }
+
+  static void processRecord(Schema recordSchema, StringBuilder stringBuilder) {
+    boolean isCommaNeeded = false;
+    for (Schema.Field field : recordSchema.getFields()) {
+      if (isCommaNeeded) {
+        stringBuilder.append(", ");
+      }
+      String fieldName = field.name();
+      stringBuilder.append(fieldName).append(" ");
+      try {
+        populateFieldsToQuery(field.schema(), stringBuilder);
+        isCommaNeeded = true;
+      } catch (Exception e) {
+        int i = stringBuilder.indexOf(fieldName);
+        if (i != -1) {
+          stringBuilder.delete(i, i + fieldName.length());
+          isCommaNeeded = false;
+        }
+      }
+    }
+  }
+
+  static String getCreateUDTTypeForNative(CassandraMapping mapping, Class persistentClass, String udtType, String fieldName) throws NoSuchFieldException {
+    StringBuilder stringBuffer = new StringBuilder();
+    Class udtClass = persistentClass.getDeclaredField(fieldName).getType();
+    UDT annotation = (UDT) udtClass.getAnnotation(UDT.class);
+    if (annotation != null) {
+      stringBuffer.append("CREATE TYPE IF NOT EXISTS ").append(mapping.getKeySpace().getName()).append(".").append(udtType).append(" (");
+      boolean isCommaNeeded = false;
+      for (java.lang.reflect.Field udtField : udtClass.getDeclaredFields()) {
+        com.datastax.driver.mapping.annotations.Field fieldAnnotation = udtField.getDeclaredAnnotation(com.datastax.driver.mapping.annotations.Field.class);
+        if (fieldAnnotation != null) {
+          if (isCommaNeeded) {
+            stringBuffer.append(", ");
+          }
+          if (!fieldAnnotation.name().isEmpty()) {
+            stringBuffer.append(fieldAnnotation.name()).append(" ");
+          } else {
+            stringBuffer.append(udtField.getName()).append(" ");
+          }
+          stringBuffer.append(dataType(udtField, null));
+          isCommaNeeded = true;
+        }
+      }
+      stringBuffer.append(")");
+    } else {
+      throw new RuntimeException("");
+    }
+    return stringBuffer.toString();
+  }
+
+  static String getCreateUDTTypeForAvro(CassandraMapping mapping, String udtType, Schema fieldSchema) {
+    StringBuilder stringBuffer = new StringBuilder();
+    stringBuffer.append("CREATE TYPE IF NOT EXISTS ").append(mapping.getKeySpace().getName()).append(".").append(udtType).append(" (");
+    CassandraQueryFactory.processRecord(fieldSchema, stringBuffer);
+    stringBuffer.append(")");
+    return stringBuffer.toString();
+  }
+
+  private static String dataType(java.lang.reflect.Field field, Type fieldType) {
+    String type;
+    if (field != null) {
+      type = field.getType().getName();
+    } else {
+      type = fieldType.getTypeName();
+    }
+    String dataType;
+    String s = type;
+    if (s.equals("java.lang.String") || s.equals("java.lang.CharSequence")) {
+      dataType = "text";
+    } else if (s.equals("int") || s.equals("java.lang.Integer")) {
+      dataType = "int";
+    } else if (s.equals("double") || s.equals("java.lang.Double")) {
+      dataType = "double";
+    } else if (s.equals("float") || s.equals("java.lang.Float")) {
+      dataType = "float";
+    } else if (s.equals("boolean") || s.equals("java.lang.Boolean")) {
+      dataType = "boolean";
+    } else if (s.equals("java.util.UUID")) {
+      dataType = "uuid";
+    } else if (s.equals("java.lang.Long")) {
+      dataType = "bigint";
+    } else if (s.equals("java.math.BigDecimal")) {
+      dataType = "decimal";
+    } else if (s.equals("java.net.InetAddress")) {
+      dataType = "inet";
+    } else if (s.equals("java.math.BigInteger")) {
+      dataType = "varint";
+    } else if (s.equals("java.nio.ByteBuffer")) {
+      dataType = "blob";
+    } else if (s.contains("Map")) {
+      ParameterizedType mapType;
+      if (field != null) {
+        mapType = (ParameterizedType) field.getGenericType();
+      } else {
+        mapType = (ParameterizedType) fieldType;
+      }
+      Type value1 = mapType.getActualTypeArguments()[0];
+      Type value2 = mapType.getActualTypeArguments()[1];
+      dataType = "map<" + dataType(null, value1) + "," + dataType(null, value2) + ">";
+    } else if (s.contains("List")) {
+      ParameterizedType listType;
+      if (field != null) {
+        listType = (ParameterizedType) field.getGenericType();
+      } else {
+        listType = (ParameterizedType) fieldType;
+      }
+      Type value = listType.getActualTypeArguments()[0];
+      dataType = "list<" + dataType(null, value) + ">";
+    } else if (s.contains("Set")) {
+      ParameterizedType setType;
+      if (field != null) {
+        setType = (ParameterizedType) field.getGenericType();
+      } else {
+        setType = (ParameterizedType) fieldType;
+      }
+      Type value = setType.getActualTypeArguments()[0];
+      dataType = "set<" + dataType(null, value) + ">";
+    } else {
+      throw new RuntimeException("Unsupported Cassandra DataType");
+    }
+    return dataType;
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/gora/blob/89683c74/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/CassandraSerializer.java
----------------------------------------------------------------------
diff --git a/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/CassandraSerializer.java b/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/CassandraSerializer.java
new file mode 100644
index 0000000..208428e
--- /dev/null
+++ b/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/CassandraSerializer.java
@@ -0,0 +1,225 @@
+/*
+ *  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.gora.cassandra.serializers;
+
+import com.datastax.driver.core.ConsistencyLevel;
+import com.datastax.driver.core.KeyspaceMetadata;
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.SimpleStatement;
+import com.datastax.driver.core.TableMetadata;
+import org.apache.gora.cassandra.bean.Field;
+import org.apache.gora.cassandra.store.CassandraClient;
+import org.apache.gora.cassandra.store.CassandraMapping;
+import org.apache.gora.cassandra.store.CassandraStore;
+import org.apache.gora.persistency.Persistent;
+import org.apache.gora.query.Query;
+import org.apache.gora.query.Result;
+import org.apache.gora.store.DataStore;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+
+/**
+ * This is the abstract Cassandra Serializer class.
+ */
+public abstract class CassandraSerializer<K, T extends Persistent> {
+
+  private static final Logger LOG = LoggerFactory.getLogger(CassandraStore.class);
+
+  protected Class<K> keyClass;
+
+  protected Class<T> persistentClass;
+
+  protected CassandraMapping mapping;
+
+  protected CassandraClient client;
+
+  protected String readConsistencyLevel;
+
+  protected String writeConsistencyLevel;
+
+  protected Map<String, String> userDefineTypeMaps;
+
+  CassandraSerializer(CassandraClient cc, Class<K> keyClass, Class<T> persistantClass, CassandraMapping mapping) {
+    this.keyClass = keyClass;
+    this.persistentClass = persistantClass;
+    this.client = cc;
+    this.mapping = mapping;
+    this.readConsistencyLevel = client.getReadConsistencyLevel();
+    this.writeConsistencyLevel = client.getWriteConsistencyLevel();
+  }
+
+  /**
+   * This method returns the Cassandra Serializer according the Cassandra serializer property.
+   *
+   * @param cc        Cassandra Client
+   * @param type      Serialization type
+   * @param dataStore Cassandra DataStore
+   * @param mapping   Cassandra Mapping
+   * @param <K>       key class
+   * @param <T>       persistent class
+   * @return Serializer
+   */
+  public static <K, T extends Persistent> CassandraSerializer getSerializer(CassandraClient cc, String type, final DataStore<K, T> dataStore, CassandraMapping mapping) {
+    CassandraStore.SerializerType serType = type == null || type.isEmpty() ? CassandraStore.SerializerType.NATIVE : CassandraStore.SerializerType.valueOf(type.toUpperCase(Locale.ENGLISH));
+    CassandraSerializer serializer;
+    switch (serType) {
+      case AVRO:
+        serializer = new AvroSerializer(cc, dataStore, mapping);
+        break;
+      case NATIVE:
+      default:
+        serializer = new NativeSerializer(cc, dataStore.getKeyClass(), dataStore.getPersistentClass(), mapping);
+    }
+    return serializer;
+  }
+
+  /**
+   * In this method persistent class been analyzed to find inner records with UDT type, this method should call in every Cassandra serialization Constructor.
+   *
+   * @throws Exception
+   */
+  protected abstract void analyzePersistent() throws Exception;
+
+
+  public void createSchema() {
+    LOG.debug("creating Cassandra keyspace {}", mapping.getKeySpace().getName());
+    this.client.getSession().execute(CassandraQueryFactory.getCreateKeySpaceQuery(mapping));
+    for (Map.Entry udtType : userDefineTypeMaps.entrySet()) {
+      LOG.debug("creating Cassandra User Define Type {}", udtType.getKey());
+      this.client.getSession().execute((String) udtType.getValue());
+    }
+    LOG.debug("creating Cassandra column family / table {}", mapping.getCoreName());
+    this.client.getSession().execute(CassandraQueryFactory.getCreateTableQuery(mapping));
+  }
+
+  public void deleteSchema() {
+    LOG.debug("dropping Cassandra table {}", mapping.getCoreName());
+    this.client.getSession().execute(CassandraQueryFactory.getDropTableQuery(mapping));
+    LOG.debug("dropping Cassandra keyspace {}", mapping.getKeySpace().getName());
+    this.client.getSession().execute(CassandraQueryFactory.getDropKeySpaceQuery(mapping));
+  }
+
+  public void close() {
+    this.client.close();
+  }
+
+  public void truncateSchema() {
+    LOG.debug("truncating Cassandra table {}", mapping.getCoreName());
+    this.client.getSession().execute(CassandraQueryFactory.getTruncateTableQuery(mapping));
+  }
+
+  public boolean schemaExists() {
+    KeyspaceMetadata keyspace = this.client.getCluster().getMetadata().getKeyspace(mapping.getKeySpace().getName());
+    if (keyspace != null) {
+      TableMetadata table = keyspace.getTable(mapping.getCoreName());
+      return table != null;
+    } else {
+      return false;
+    }
+  }
+
+  protected String[] getFields() {
+    List<String> fields = new ArrayList<>();
+    for (Field field : mapping.getFieldList()) {
+      fields.add(field.getFieldName());
+    }
+    return fields.toArray(new String[0]);
+  }
+
+  /**
+   * Inserts the persistent Object
+   *
+   * @param key   key value
+   * @param value persistent value
+   */
+  public abstract void put(K key, T value);
+
+  /**
+   * Retrieves the persistent value according to the key
+   *
+   * @param key key value
+   * @return persistent value
+   */
+  public abstract T get(K key);
+
+  /**
+   * Deletes persistent value according to the key
+   *
+   * @param key key value
+   * @return isDeleted
+   */
+  public abstract boolean delete(K key);
+
+  /**
+   * Retrieves the persistent value according to the key and fields
+   *
+   * @param key    key value
+   * @param fields fields
+   * @return persistent value
+   */
+  public abstract T get(K key, String[] fields);
+
+  /**
+   * Executes the given query and returns the results.
+   *
+   * @param dataStore Cassandra data store
+   * @param query     Cassandra Query
+   * @return Cassandra Result
+   */
+  public abstract Result<K, T> execute(DataStore<K, T> dataStore, Query<K, T> query);
+
+  /**
+   * Update the persistent objects
+   *
+   * @param query Cassandra Query
+   * @return isUpdated
+   */
+  public abstract boolean updateByQuery(Query query);
+
+  public long deleteByQuery(Query query) {
+    List<Object> objectArrayList = new ArrayList<>();
+    if (query.getKey() == null && query.getEndKey() == null && query.getStartKey() == null) {
+      if (query.getFields() == null) {
+        client.getSession().execute(CassandraQueryFactory.getTruncateTableQuery(mapping));
+      } else {
+        LOG.error("Delete by Query is not supported for the Queries which didn't specify Query keys with fields.");
+      }
+    } else {
+      String cqlQuery = CassandraQueryFactory.getDeleteByQuery(mapping, query, objectArrayList);
+      ResultSet results;
+      SimpleStatement statement;
+      if (objectArrayList.size() == 0) {
+        statement = new SimpleStatement(cqlQuery);
+      } else {
+        statement = new SimpleStatement(cqlQuery, objectArrayList.toArray());
+      }
+      if (writeConsistencyLevel != null) {
+        statement.setConsistencyLevel(ConsistencyLevel.valueOf(writeConsistencyLevel));
+      }
+      results = client.getSession().execute(statement);
+      LOG.debug("Delete by Query was applied : " + results.wasApplied());
+    }
+    LOG.info("Delete By Query method doesn't return the deleted element count.");
+    return 0;
+  }
+
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/gora/blob/89683c74/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/NativeSerializer.java
----------------------------------------------------------------------
diff --git a/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/NativeSerializer.java b/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/NativeSerializer.java
new file mode 100644
index 0000000..bf28ee0
--- /dev/null
+++ b/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/NativeSerializer.java
@@ -0,0 +1,243 @@
+/*
+ *  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.gora.cassandra.serializers;
+
+import com.datastax.driver.core.ConsistencyLevel;
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.SimpleStatement;
+import com.datastax.driver.mapping.Mapper;
+import com.datastax.driver.mapping.MappingManager;
+import com.datastax.driver.mapping.Result;
+import org.apache.commons.lang.ArrayUtils;
+import org.apache.gora.cassandra.bean.Field;
+import org.apache.gora.cassandra.query.CassandraResultSet;
+import org.apache.gora.cassandra.store.CassandraClient;
+import org.apache.gora.cassandra.store.CassandraMapping;
+import org.apache.gora.persistency.Persistent;
+import org.apache.gora.query.Query;
+import org.apache.gora.store.DataStore;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+
+/**
+ * This Class contains the operation relates to Native Serialization.
+ */
+class NativeSerializer<K, T extends Persistent> extends CassandraSerializer {
+
+  private static final Logger LOG = LoggerFactory.getLogger(NativeSerializer.class);
+
+  private Mapper<T> mapper;
+
+  NativeSerializer(CassandraClient cassandraClient, Class<K> keyClass, Class<T> persistentClass, CassandraMapping mapping) {
+    super(cassandraClient, keyClass, persistentClass, mapping);
+    try {
+      analyzePersistent();
+    } catch (Exception e) {
+      throw new RuntimeException("Error occurred while analyzing the persistent class, :" + e.getMessage());
+    }
+    this.createSchema();
+    MappingManager mappingManager = new MappingManager(cassandraClient.getSession());
+    mapper = mappingManager.mapper(persistentClass);
+    if (cassandraClient.getWriteConsistencyLevel() != null) {
+      mapper.setDefaultDeleteOptions(Mapper.Option.consistencyLevel(ConsistencyLevel.valueOf(cassandraClient.getWriteConsistencyLevel())));
+      mapper.setDefaultSaveOptions(Mapper.Option.consistencyLevel(ConsistencyLevel.valueOf(cassandraClient.getWriteConsistencyLevel())));
+    }
+    if (cassandraClient.getReadConsistencyLevel() != null) {
+      mapper.setDefaultGetOptions(Mapper.Option.consistencyLevel(ConsistencyLevel.valueOf(cassandraClient.getReadConsistencyLevel())));
+    }
+  }
+
+  /**
+   * {@inheritDoc}
+   *
+   * @param key
+   * @param value
+   */
+  @Override
+  public void put(Object key, Persistent value) {
+    LOG.debug("Object is saved with key : {} and value : {}", key, value);
+    mapper.save((T) value);
+  }
+
+  /**
+   * {@inheritDoc}
+   *
+   * @param key
+   * @return
+   */
+  @Override
+  public T get(Object key) {
+    T object = mapper.get(key);
+    if (object != null) {
+      LOG.debug("Object is found for key : {}", key);
+    } else {
+      LOG.debug("Object is not found for key : {}", key);
+    }
+    return object;
+  }
+
+  /**
+   * {@inheritDoc}
+   *
+   * @param key
+   * @return
+   */
+  @Override
+  public boolean delete(Object key) {
+    LOG.debug("Object is deleted for key : {}", key);
+    mapper.delete(key);
+    return true;
+  }
+
+  /**
+   * {@inheritDoc}
+   *
+   * @param key
+   * @param fields
+   * @return
+   */
+  @Override
+  public Persistent get(Object key, String[] fields) {
+    if (fields == null) {
+      fields = getFields();
+    }
+    String cqlQuery = CassandraQueryFactory.getSelectObjectWithFieldsQuery(mapping, fields);
+    SimpleStatement statement = new SimpleStatement(cqlQuery, key);
+    if (readConsistencyLevel != null) {
+      statement.setConsistencyLevel(ConsistencyLevel.valueOf(readConsistencyLevel));
+    }
+    ResultSet results = client.getSession().execute(statement);
+    Result<T> objects = mapper.map(results);
+    List<T> objectList = objects.all();
+    if (objectList != null) {
+      LOG.debug("Object is found for key : {}", key);
+      return objectList.get(0);
+    }
+    LOG.debug("Object is not found for key : {}", key);
+    return null;
+  }
+
+  /**
+   * {@inheritDoc}
+   *
+   * @throws Exception
+   */
+  @Override
+  protected void analyzePersistent() throws Exception {
+    userDefineTypeMaps = new HashMap<>();
+    for (Field field : mapping.getFieldList()) {
+      String fieldType = field.getType();
+      if (fieldType.contains("frozen")) {
+        String udtType = fieldType.substring(fieldType.indexOf("<") + 1, fieldType.indexOf(">"));
+        String createQuery = CassandraQueryFactory.getCreateUDTTypeForNative(mapping, persistentClass, udtType, field.getFieldName());
+        userDefineTypeMaps.put(udtType, createQuery);
+      }
+    }
+  }
+
+  /**
+   * {@inheritDoc}
+   *
+   * @param query
+   * @return
+   */
+  @Override
+  public boolean updateByQuery(Query query) {
+    List<Object> objectArrayList = new ArrayList<>();
+    String cqlQuery = CassandraQueryFactory.getUpdateByQueryForNative(mapping, query, objectArrayList);
+    ResultSet results;
+    SimpleStatement statement;
+    if (objectArrayList.size() == 0) {
+      statement = new SimpleStatement(cqlQuery);
+    } else {
+      statement = new SimpleStatement(cqlQuery, objectArrayList.toArray());
+    }
+    if (writeConsistencyLevel != null) {
+      statement.setConsistencyLevel(ConsistencyLevel.valueOf(writeConsistencyLevel));
+    }
+    results = client.getSession().execute(statement);
+    return results.wasApplied();
+  }
+
+  /**
+   * {@inheritDoc}
+   *
+   * @param dataStore
+   * @param query
+   * @return
+   */
+  @Override
+  public org.apache.gora.query.Result execute(DataStore dataStore, Query query) {
+    List<Object> objectArrayList = new ArrayList<>();
+    String[] fields = query.getFields();
+    if (fields != null) {
+      fields = (String[]) ArrayUtils.addAll(fields, mapping.getAllKeys());
+    } else {
+      fields = mapping.getAllFieldsIncludingKeys();
+    }
+    CassandraResultSet<K, T> cassandraResult = new CassandraResultSet<>(dataStore, query);
+    String cqlQuery = CassandraQueryFactory.getExecuteQuery(mapping, query, objectArrayList, fields);
+    ResultSet results;
+    if (objectArrayList.size() == 0) {
+      results = client.getSession().execute(cqlQuery);
+    } else {
+      results = client.getSession().execute(cqlQuery, objectArrayList.toArray());
+    }
+    Result<T> objects = mapper.map(results);
+    Iterator iterator = objects.iterator();
+    while (iterator.hasNext()) {
+      T result = (T) iterator.next();
+      K key = getKey(result);
+      cassandraResult.addResultElement(key, result);
+    }
+    return cassandraResult;
+  }
+
+  private K getKey(T object) {
+    String keyField = null;
+    for (Field field : mapping.getFieldList()) {
+      boolean isPrimaryKey = Boolean.parseBoolean(field.getProperty("primarykey"));
+      if (isPrimaryKey) {
+        keyField = field.getFieldName();
+        break;
+      }
+    }
+    K key;
+    Method keyMethod = null;
+    try {
+      for (Method method : this.persistentClass.getMethods()) {
+        if (method.getName().equalsIgnoreCase("get" + keyField)) {
+          keyMethod = method;
+        }
+      }
+      key = (K) keyMethod.invoke(object);
+    } catch (Exception e) {
+      try {
+        key = (K) this.persistentClass.getField(keyField).get(object);
+      } catch (Exception e1) {
+        throw new RuntimeException("Field" + keyField + " is not accessible in " + persistentClass + " : " + e1.getMessage());
+      }
+    }
+    return key;
+  }
+}

http://git-wip-us.apache.org/repos/asf/gora/blob/89683c74/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/package-info.java
----------------------------------------------------------------------
diff --git a/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/package-info.java b/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/package-info.java
new file mode 100644
index 0000000..ce1e3e7
--- /dev/null
+++ b/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/package-info.java
@@ -0,0 +1,21 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.
+ */
+
+/**
+ * This package contains Cassandra store related util classes for serializer.
+ */
+package org.apache.gora.cassandra.serializers;
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/gora/blob/89683c74/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/CassandraClient.java
----------------------------------------------------------------------
diff --git a/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/CassandraClient.java b/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/CassandraClient.java
new file mode 100644
index 0000000..f672884
--- /dev/null
+++ b/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/CassandraClient.java
@@ -0,0 +1,535 @@
+/*
+ *  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.gora.cassandra.store;
+
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.ConsistencyLevel;
+import com.datastax.driver.core.DataType;
+import com.datastax.driver.core.HostDistance;
+import com.datastax.driver.core.PoolingOptions;
+import com.datastax.driver.core.ProtocolOptions;
+import com.datastax.driver.core.ProtocolVersion;
+import com.datastax.driver.core.QueryOptions;
+import com.datastax.driver.core.Session;
+import com.datastax.driver.core.SocketOptions;
+import com.datastax.driver.core.TypeCodec;
+import com.datastax.driver.core.policies.ConstantReconnectionPolicy;
+import com.datastax.driver.core.policies.DCAwareRoundRobinPolicy;
+import com.datastax.driver.core.policies.DefaultRetryPolicy;
+import com.datastax.driver.core.policies.DowngradingConsistencyRetryPolicy;
+import com.datastax.driver.core.policies.ExponentialReconnectionPolicy;
+import com.datastax.driver.core.policies.FallthroughRetryPolicy;
+import com.datastax.driver.core.policies.LatencyAwarePolicy;
+import com.datastax.driver.core.policies.LoggingRetryPolicy;
+import com.datastax.driver.core.policies.RoundRobinPolicy;
+import com.datastax.driver.core.policies.TokenAwarePolicy;
+import com.datastax.driver.extras.codecs.arrays.DoubleArrayCodec;
+import com.datastax.driver.extras.codecs.arrays.FloatArrayCodec;
+import com.datastax.driver.extras.codecs.arrays.IntArrayCodec;
+import com.datastax.driver.extras.codecs.arrays.LongArrayCodec;
+import com.datastax.driver.extras.codecs.arrays.ObjectArrayCodec;
+import com.datastax.driver.extras.codecs.date.SimpleDateCodec;
+import com.datastax.driver.extras.codecs.date.SimpleTimestampCodec;
+import com.datastax.driver.extras.codecs.jdk8.OptionalCodec;
+import org.apache.gora.cassandra.bean.Field;
+import org.jdom.Document;
+import org.jdom.Element;
+import org.jdom.JDOMException;
+import org.jdom.input.SAXBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Locale;
+import java.util.Properties;
+
+/**
+ * This class provides the Cassandra Client Connection.
+ * Initialize the Cassandra Connection according to the Properties.
+ */
+public class CassandraClient {
+
+  private static final Logger LOG = LoggerFactory.getLogger(CassandraClient.class);
+
+  private Cluster cluster;
+
+  private Session session;
+
+  private CassandraMapping mapping;
+
+  private String readConsistencyLevel;
+
+  private String writeConsistencyLevel;
+
+  public Session getSession() {
+    return session;
+  }
+
+  public Cluster getCluster() {
+    return cluster;
+  }
+
+  void initialize(Properties properties, CassandraMapping mapping) throws Exception {
+    Cluster.Builder builder = Cluster.builder();
+    List<String> codecs = readCustomCodec(properties);
+    builder = populateSettings(builder, properties);
+    this.mapping = mapping;
+    this.cluster = builder.build();
+    if (codecs != null) {
+      registerCustomCodecs(codecs);
+    }
+    readConsistencyLevel = properties.getProperty(CassandraStoreParameters.READ_CONSISTENCY_LEVEL);
+    writeConsistencyLevel = properties.getProperty(CassandraStoreParameters.WRITE_CONSISTENCY_LEVEL);
+    registerOptionalCodecs();
+    this.session = this.cluster.connect();
+  }
+
+  private void registerOptionalCodecs() {
+    // Optional Codecs for natives
+    this.cluster.getConfiguration().getCodecRegistry().register(new OptionalCodec<>(TypeCodec.ascii()));
+    this.cluster.getConfiguration().getCodecRegistry().register(new OptionalCodec<>(TypeCodec.bigint()));
+    this.cluster.getConfiguration().getCodecRegistry().register(new OptionalCodec<>(TypeCodec.blob()));
+    this.cluster.getConfiguration().getCodecRegistry().register(new OptionalCodec<>(TypeCodec.cboolean()));
+    this.cluster.getConfiguration().getCodecRegistry().register(new OptionalCodec<>(TypeCodec.cdouble()));
+    this.cluster.getConfiguration().getCodecRegistry().register(new OptionalCodec<>(TypeCodec.cfloat()));
+    this.cluster.getConfiguration().getCodecRegistry().register(new OptionalCodec<>(TypeCodec.cint()));
+    this.cluster.getConfiguration().getCodecRegistry().register(new OptionalCodec<>(TypeCodec.counter()));
+    this.cluster.getConfiguration().getCodecRegistry().register(new OptionalCodec<>(TypeCodec.date()));
+    this.cluster.getConfiguration().getCodecRegistry().register(new OptionalCodec<>(TypeCodec.decimal()));
+    this.cluster.getConfiguration().getCodecRegistry().register(new OptionalCodec<>(TypeCodec.inet()));
+    this.cluster.getConfiguration().getCodecRegistry().register(new OptionalCodec<>(TypeCodec.smallInt()));
+    this.cluster.getConfiguration().getCodecRegistry().register(new OptionalCodec<>(TypeCodec.time()));
+    this.cluster.getConfiguration().getCodecRegistry().register(new OptionalCodec<>(TypeCodec.timestamp()));
+    this.cluster.getConfiguration().getCodecRegistry().register(new OptionalCodec<>(TypeCodec.timeUUID()));
+    this.cluster.getConfiguration().getCodecRegistry().register(new OptionalCodec<>(TypeCodec.tinyInt()));
+    this.cluster.getConfiguration().getCodecRegistry().register(new OptionalCodec<>(TypeCodec.varint()));
+    this.cluster.getConfiguration().getCodecRegistry().register(new OptionalCodec<>(TypeCodec.varchar()));
+    this.cluster.getConfiguration().getCodecRegistry().register(new OptionalCodec<>(TypeCodec.uuid()));
+    // Optional Array Codecs
+    this.cluster.getConfiguration().getCodecRegistry().register(new IntArrayCodec());
+    this.cluster.getConfiguration().getCodecRegistry().register(new DoubleArrayCodec());
+    this.cluster.getConfiguration().getCodecRegistry().register(new FloatArrayCodec());
+    this.cluster.getConfiguration().getCodecRegistry().register(new LongArrayCodec());
+    this.cluster.getConfiguration().getCodecRegistry().register(new ObjectArrayCodec<>(
+            DataType.list(DataType.varchar()),
+            String[].class,
+            TypeCodec.varchar()));
+    // Optional Time Codecs
+    this.cluster.getConfiguration().getCodecRegistry().register(new SimpleDateCodec());
+    this.cluster.getConfiguration().getCodecRegistry().register(new SimpleTimestampCodec());
+
+    for (Field field : this.mapping.getFieldList()) {
+      String columnType = field.getType().toLowerCase(Locale.ENGLISH);
+      //http://docs.datastax.com/en/cql/3.3/cql/cql_reference/cql_data_types_c.html
+      if (columnType.contains("list")) {
+        columnType = columnType.substring(columnType.indexOf("<") + 1, columnType.indexOf(">"));
+        this.cluster.getConfiguration().getCodecRegistry().register(new OptionalCodec<>(TypeCodec.list(getTypeCodec(columnType))));
+      } else if (columnType.contains("set")) {
+        columnType = columnType.substring(columnType.indexOf("<") + 1, columnType.indexOf(">"));
+        this.cluster.getConfiguration().getCodecRegistry().register(new OptionalCodec<>(TypeCodec.set(getTypeCodec(columnType))));
+      } else if (columnType.contains("map")) {
+        String[] columnTypes = columnType.substring(columnType.indexOf("<") + 1, columnType.indexOf(">")).split(",");
+        this.cluster.getConfiguration().getCodecRegistry().register(new OptionalCodec<>(TypeCodec.map(TypeCodec.set(getTypeCodec(columnTypes[0])), TypeCodec.set(getTypeCodec(columnTypes[1])))));
+      }
+    }
+  }
+
+  private TypeCodec getTypeCodec(String columnType) {
+    TypeCodec typeCodec;
+    switch (columnType) {
+      case "ascii":
+        typeCodec = TypeCodec.ascii();
+        break;
+      case "bigint":
+        typeCodec = TypeCodec.bigint();
+        break;
+      case "blob":
+        typeCodec = TypeCodec.blob();
+        break;
+      case "boolean":
+        typeCodec = TypeCodec.cboolean();
+        break;
+      case "counter":
+        typeCodec = TypeCodec.counter();
+        break;
+      case "date":
+        typeCodec = TypeCodec.date();
+        break;
+      case "decimal":
+        typeCodec = TypeCodec.decimal();
+        break;
+      case "double":
+        typeCodec = TypeCodec.cdouble();
+        break;
+      case "float":
+        typeCodec = TypeCodec.cfloat();
+        break;
+      case "inet":
+        typeCodec = TypeCodec.inet();
+        break;
+      case "int":
+        typeCodec = TypeCodec.cint();
+        break;
+      case "smallint":
+        typeCodec = TypeCodec.smallInt();
+        break;
+      case "time":
+        typeCodec = TypeCodec.time();
+        break;
+      case "timestamp":
+        typeCodec = TypeCodec.timestamp();
+        break;
+      case "timeuuid":
+        typeCodec = TypeCodec.timeUUID();
+        break;
+      case "tinyint":
+        typeCodec = TypeCodec.tinyInt();
+        break;
+      case "uuid":
+        typeCodec = TypeCodec.uuid();
+        break;
+      case "varint":
+        typeCodec = TypeCodec.varint();
+        break;
+      case "varchar":
+      case "text":
+        typeCodec = TypeCodec.varchar();
+        break;
+      default:
+        LOG.error("Unsupported Cassandra datatype: {} ", columnType);
+        throw new RuntimeException("Unsupported Cassandra datatype: " + columnType);
+    }
+    return typeCodec;
+  }
+
+  private Cluster.Builder populateSettings(Cluster.Builder builder, Properties properties) {
+    String serversParam = properties.getProperty(CassandraStoreParameters.CASSANDRA_SERVERS);
+    String[] servers = serversParam.split(",");
+    for (String server : servers) {
+      builder = builder.addContactPoint(server);
+    }
+    String portProp = properties.getProperty(CassandraStoreParameters.PORT);
+    if (portProp != null) {
+      builder = builder.withPort(Integer.parseInt(portProp));
+    }
+    String clusterNameProp = properties.getProperty(CassandraStoreParameters.CLUSTER_NAME);
+    if (clusterNameProp != null) {
+      builder = builder.withClusterName(clusterNameProp);
+    }
+    String compressionProp = properties.getProperty(CassandraStoreParameters.COMPRESSION);
+    if (compressionProp != null) {
+      builder = builder.withCompression(ProtocolOptions.Compression.valueOf(compressionProp));
+    }
+    builder = this.populateCredentials(properties, builder);
+    builder = this.populateLoadBalancingProp(properties, builder);
+    String enableJMXProp = properties.getProperty(CassandraStoreParameters.ENABLE_JMX_REPORTING);
+    if (!Boolean.parseBoolean(enableJMXProp)) {
+      builder = builder.withoutJMXReporting();
+    }
+    String enableMetricsProp = properties.getProperty(CassandraStoreParameters.ENABLE_METRICS);
+    if (!Boolean.parseBoolean(enableMetricsProp)) {
+      builder = builder.withoutMetrics();
+    }
+    builder = this.populatePoolingSettings(properties, builder);
+    String versionProp = properties.getProperty(CassandraStoreParameters.PROTOCOL_VERSION);
+    if (versionProp != null) {
+      builder = builder.withProtocolVersion(ProtocolVersion.fromInt(Integer.parseInt(versionProp)));
+    }
+    builder = this.populateQueryOptions(properties, builder);
+    builder = this.populateReconnectPolicy(properties, builder);
+    builder = this.populateRetrytPolicy(properties, builder);
+    builder = this.populateSocketOptions(properties, builder);
+    String enableSSLProp = properties.getProperty(CassandraStoreParameters.ENABLE_SSL);
+    if (enableSSLProp != null) {
+      if (Boolean.parseBoolean(enableSSLProp)) {
+        builder = builder.withSSL();
+      }
+    }
+    return builder;
+  }
+
+  private Cluster.Builder populateLoadBalancingProp(Properties properties, Cluster.Builder builder) {
+    String loadBalancingProp = properties.getProperty(CassandraStoreParameters.LOAD_BALANCING_POLICY);
+    if (loadBalancingProp != null) {
+      switch (loadBalancingProp) {
+        case "LatencyAwareRoundRobinPolicy":
+          builder = builder.withLoadBalancingPolicy(LatencyAwarePolicy.builder(new RoundRobinPolicy()).build());
+          break;
+        case "RoundRobinPolicy":
+          builder = builder.withLoadBalancingPolicy(new RoundRobinPolicy());
+          break;
+        case "DCAwareRoundRobinPolicy": {
+          String dataCenter = properties.getProperty(CassandraStoreParameters.DATA_CENTER);
+          boolean allowRemoteDCsForLocalConsistencyLevel = Boolean.parseBoolean(
+                  properties.getProperty(CassandraStoreParameters.ALLOW_REMOTE_DCS_FOR_LOCAL_CONSISTENCY_LEVEL));
+          if (dataCenter != null && !dataCenter.isEmpty()) {
+            if (allowRemoteDCsForLocalConsistencyLevel) {
+              builder = builder.withLoadBalancingPolicy(
+                      DCAwareRoundRobinPolicy.builder().withLocalDc(dataCenter)
+                              .allowRemoteDCsForLocalConsistencyLevel().build());
+            } else {
+              builder = builder.withLoadBalancingPolicy(
+                      DCAwareRoundRobinPolicy.builder().withLocalDc(dataCenter).build());
+            }
+          } else {
+            if (allowRemoteDCsForLocalConsistencyLevel) {
+              builder = builder.withLoadBalancingPolicy(
+                      (DCAwareRoundRobinPolicy.builder().allowRemoteDCsForLocalConsistencyLevel().build()));
+            } else {
+              builder = builder.withLoadBalancingPolicy((DCAwareRoundRobinPolicy.builder().build()));
+            }
+          }
+          break;
+        }
+        case "TokenAwareRoundRobinPolicy":
+          builder = builder.withLoadBalancingPolicy(new TokenAwarePolicy(new RoundRobinPolicy()));
+          break;
+        case "TokenAwareDCAwareRoundRobinPolicy": {
+          String dataCenter = properties.getProperty(CassandraStoreParameters.DATA_CENTER);
+          boolean allowRemoteDCsForLocalConsistencyLevel = Boolean.parseBoolean(
+                  properties.getProperty(CassandraStoreParameters.ALLOW_REMOTE_DCS_FOR_LOCAL_CONSISTENCY_LEVEL));
+          if (dataCenter != null && !dataCenter.isEmpty()) {
+            if (allowRemoteDCsForLocalConsistencyLevel) {
+              builder = builder.withLoadBalancingPolicy(new TokenAwarePolicy(
+                      DCAwareRoundRobinPolicy.builder().withLocalDc(dataCenter)
+                              .allowRemoteDCsForLocalConsistencyLevel().build()));
+            } else {
+              builder = builder.withLoadBalancingPolicy(new TokenAwarePolicy(
+                      DCAwareRoundRobinPolicy.builder().withLocalDc(dataCenter).build()));
+            }
+          } else {
+            if (allowRemoteDCsForLocalConsistencyLevel) {
+              builder = builder.withLoadBalancingPolicy(new TokenAwarePolicy(
+                      DCAwareRoundRobinPolicy.builder().allowRemoteDCsForLocalConsistencyLevel().build()));
+            } else {
+              builder = builder.withLoadBalancingPolicy(
+                      new TokenAwarePolicy(DCAwareRoundRobinPolicy.builder().build()));
+            }
+          }
+          break;
+        }
+        default:
+          LOG.error("Unsupported Cassandra load balancing  policy: {} ", loadBalancingProp);
+          break;
+      }
+    }
+    return builder;
+  }
+
+  private Cluster.Builder populateCredentials(Properties properties, Cluster.Builder builder) {
+    String usernameProp = properties.getProperty(CassandraStoreParameters.USERNAME);
+    String passwordProp = properties.getProperty(CassandraStoreParameters.PASSWORD);
+    if (usernameProp != null) {
+      builder = builder.withCredentials(usernameProp, passwordProp);
+    }
+    return builder;
+  }
+
+  private Cluster.Builder populatePoolingSettings(Properties properties, Cluster.Builder builder) {
+    String localCoreConnectionsPerHost = properties.getProperty(CassandraStoreParameters.LOCAL_CORE_CONNECTIONS_PER_HOST);
+    String remoteCoreConnectionsPerHost = properties.getProperty(CassandraStoreParameters.REMOTE_CORE_CONNECTIONS_PER_HOST);
+    String localMaxConnectionsPerHost = properties.getProperty(CassandraStoreParameters.LOCAL_MAX_CONNECTIONS_PER_HOST);
+    String remoteMaxConnectionsPerHost = properties.getProperty(CassandraStoreParameters.REMOTE_MAX_CONNECTIONS_PER_HOST);
+    String localNewConnectionThreshold = properties.getProperty(CassandraStoreParameters.LOCAL_NEW_CONNECTION_THRESHOLD);
+    String remoteNewConnectionThreshold = properties.getProperty(CassandraStoreParameters.REMOTE_NEW_CONNECTION_THRESHOLD);
+    String localMaxRequestsPerConnection = properties.getProperty(CassandraStoreParameters.LOCAL_MAX_REQUESTS_PER_CONNECTION);
+    String remoteMaxRequestsPerConnection = properties.getProperty(CassandraStoreParameters.REMOTE_MAX_REQUESTS_PER_CONNECTION);
+    PoolingOptions options = new PoolingOptions();
+    if (localCoreConnectionsPerHost != null) {
+      options.setCoreConnectionsPerHost(HostDistance.LOCAL, Integer.parseInt(localCoreConnectionsPerHost));
+    }
+    if (remoteCoreConnectionsPerHost != null) {
+      options.setCoreConnectionsPerHost(HostDistance.REMOTE, Integer.parseInt(remoteCoreConnectionsPerHost));
+    }
+    if (localMaxConnectionsPerHost != null) {
+      options.setMaxConnectionsPerHost(HostDistance.LOCAL, Integer.parseInt(localMaxConnectionsPerHost));
+    }
+    if (remoteMaxConnectionsPerHost != null) {
+      options.setMaxConnectionsPerHost(HostDistance.REMOTE, Integer.parseInt(remoteMaxConnectionsPerHost));
+    }
+    if (localNewConnectionThreshold != null) {
+      options.setNewConnectionThreshold(HostDistance.LOCAL, Integer.parseInt(localNewConnectionThreshold));
+    }
+    if (remoteNewConnectionThreshold != null) {
+      options.setNewConnectionThreshold(HostDistance.REMOTE, Integer.parseInt(remoteNewConnectionThreshold));
+    }
+    if (localMaxRequestsPerConnection != null) {
+      options.setMaxRequestsPerConnection(HostDistance.LOCAL, Integer.parseInt(localMaxRequestsPerConnection));
+    }
+    if (remoteMaxRequestsPerConnection != null) {
+      options.setMaxRequestsPerConnection(HostDistance.REMOTE, Integer.parseInt(remoteMaxRequestsPerConnection));
+    }
+    builder = builder.withPoolingOptions(options);
+    return builder;
+  }
+
+  private Cluster.Builder populateQueryOptions(Properties properties, Cluster.Builder builder) {
+    String consistencyLevelProp = properties.getProperty(CassandraStoreParameters.CONSISTENCY_LEVEL);
+    String serialConsistencyLevelProp = properties.getProperty(CassandraStoreParameters.SERIAL_CONSISTENCY_LEVEL);
+    String fetchSize = properties.getProperty(CassandraStoreParameters.FETCH_SIZE);
+    QueryOptions options = new QueryOptions();
+    if (consistencyLevelProp != null) {
+      options.setConsistencyLevel(ConsistencyLevel.valueOf(consistencyLevelProp));
+    }
+    if (serialConsistencyLevelProp != null) {
+      options.setSerialConsistencyLevel(ConsistencyLevel.valueOf(serialConsistencyLevelProp));
+    }
+    if (fetchSize != null) {
+      options.setFetchSize(Integer.parseInt(fetchSize));
+    }
+    return builder.withQueryOptions(options);
+  }
+
+  private Cluster.Builder populateReconnectPolicy(Properties properties, Cluster.Builder builder) {
+    String reconnectionPolicy = properties.getProperty(CassandraStoreParameters.RECONNECTION_POLICY);
+    if (reconnectionPolicy != null) {
+      switch (reconnectionPolicy) {
+        case "ConstantReconnectionPolicy": {
+          String constantReconnectionPolicyDelay = properties.getProperty(CassandraStoreParameters.CONSTANT_RECONNECTION_POLICY_DELAY);
+          ConstantReconnectionPolicy policy = new ConstantReconnectionPolicy(Long.parseLong(constantReconnectionPolicyDelay));
+          builder = builder.withReconnectionPolicy(policy);
+          break;
+        }
+        case "ExponentialReconnectionPolicy": {
+          String exponentialReconnectionPolicyBaseDelay = properties.getProperty(CassandraStoreParameters.EXPONENTIAL_RECONNECTION_POLICY_BASE_DELAY);
+          String exponentialReconnectionPolicyMaxDelay = properties.getProperty(CassandraStoreParameters.EXPONENTIAL_RECONNECTION_POLICY_MAX_DELAY);
+
+          ExponentialReconnectionPolicy policy = new ExponentialReconnectionPolicy(Long.parseLong(exponentialReconnectionPolicyBaseDelay),
+                  Long.parseLong(exponentialReconnectionPolicyMaxDelay));
+          builder = builder.withReconnectionPolicy(policy);
+          break;
+        }
+        default:
+          LOG.error("Unsupported reconnection policy : {} ", reconnectionPolicy);
+      }
+    }
+    return builder;
+  }
+
+  private Cluster.Builder populateRetrytPolicy(Properties properties, Cluster.Builder builder) {
+    String retryPolicy = properties.getProperty(CassandraStoreParameters.RETRY_POLICY);
+    if (retryPolicy != null) {
+      switch (retryPolicy) {
+        case "DefaultRetryPolicy":
+          builder = builder.withRetryPolicy(DefaultRetryPolicy.INSTANCE);
+          break;
+        case "DowngradingConsistencyRetryPolicy":
+          builder = builder.withRetryPolicy(DowngradingConsistencyRetryPolicy.INSTANCE);
+          break;
+        case "FallthroughRetryPolicy":
+          builder = builder.withRetryPolicy(FallthroughRetryPolicy.INSTANCE);
+          break;
+        case "LoggingDefaultRetryPolicy":
+          builder = builder.withRetryPolicy(new LoggingRetryPolicy(DefaultRetryPolicy.INSTANCE));
+          break;
+        case "LoggingDowngradingConsistencyRetryPolicy":
+          builder = builder.withRetryPolicy(new LoggingRetryPolicy(DowngradingConsistencyRetryPolicy.INSTANCE));
+          break;
+        case "LoggingFallthroughRetryPolicy":
+          builder = builder.withRetryPolicy(new LoggingRetryPolicy(FallthroughRetryPolicy.INSTANCE));
+          break;
+        default:
+          LOG.error("Unsupported retry policy : {} ", retryPolicy);
+          break;
+      }
+    }
+    return builder;
+  }
+
+  private Cluster.Builder populateSocketOptions(Properties properties, Cluster.Builder builder) {
+    String connectionTimeoutMillisProp = properties.getProperty(CassandraStoreParameters.CONNECTION_TIMEOUT_MILLIS);
+    String keepAliveProp = properties.getProperty(CassandraStoreParameters.KEEP_ALIVE);
+    String readTimeoutMillisProp = properties.getProperty(CassandraStoreParameters.READ_TIMEOUT_MILLIS);
+    String receiveBufferSizeProp = properties.getProperty(CassandraStoreParameters.RECEIVER_BUFFER_SIZE);
+    String reuseAddress = properties.getProperty(CassandraStoreParameters.REUSE_ADDRESS);
+    String sendBufferSize = properties.getProperty(CassandraStoreParameters.SEND_BUFFER_SIZE);
+    String soLinger = properties.getProperty(CassandraStoreParameters.SO_LINGER);
+    String tcpNoDelay = properties.getProperty(CassandraStoreParameters.TCP_NODELAY);
+    SocketOptions options = new SocketOptions();
+    if (connectionTimeoutMillisProp != null) {
+      options.setConnectTimeoutMillis(Integer.parseInt(connectionTimeoutMillisProp));
+    }
+    if (keepAliveProp != null) {
+      options.setKeepAlive(Boolean.parseBoolean(keepAliveProp));
+    }
+    if (readTimeoutMillisProp != null) {
+      options.setReadTimeoutMillis(Integer.parseInt(readTimeoutMillisProp));
+    }
+    if (receiveBufferSizeProp != null) {
+      options.setReceiveBufferSize(Integer.parseInt(receiveBufferSizeProp));
+    }
+    if (reuseAddress != null) {
+      options.setReuseAddress(Boolean.parseBoolean(reuseAddress));
+    }
+    if (sendBufferSize != null) {
+      options.setSendBufferSize(Integer.parseInt(sendBufferSize));
+    }
+    if (soLinger != null) {
+      options.setSoLinger(Integer.parseInt(soLinger));
+    }
+    if (tcpNoDelay != null) {
+      options.setTcpNoDelay(Boolean.parseBoolean(tcpNoDelay));
+    }
+    return builder.withSocketOptions(options);
+  }
+
+  private List<String> readCustomCodec(Properties properties) throws JDOMException, IOException {
+    String filename = properties.getProperty(CassandraStoreParameters.CUSTOM_CODEC_FILE);
+    if (filename != null) {
+      List<String> codecs = new ArrayList<>();
+      SAXBuilder builder = new SAXBuilder();
+      Document doc = builder.build(getClass().getClassLoader().getResourceAsStream(filename));
+      List<Element> codecElementList = doc.getRootElement().getChildren("codec");
+      for (Element codec : codecElementList) {
+        codecs.add(codec.getValue());
+      }
+      return codecs;
+    }
+    return null;
+  }
+
+  /**
+   * This method returns configured read consistency level.
+   * @return read Consistency level
+   */
+  public String getReadConsistencyLevel() {
+    return readConsistencyLevel;
+  }
+
+  /**
+   * This method returns configured write consistency level.
+   * @return write Consistency level
+   */
+  public String getWriteConsistencyLevel() {
+    return writeConsistencyLevel;
+  }
+
+
+  public void close() {
+    this.session.close();
+    this.cluster.close();
+  }
+
+  private void registerCustomCodecs(List<String> codecs) throws Exception {
+    for (String codec : codecs) {
+      this.cluster.getConfiguration().getCodecRegistry().register((TypeCodec<?>) Class.forName(codec).newInstance());
+    }
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/gora/blob/89683c74/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/CassandraMapping.java
----------------------------------------------------------------------
diff --git a/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/CassandraMapping.java b/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/CassandraMapping.java
new file mode 100644
index 0000000..7d5ac05
--- /dev/null
+++ b/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/CassandraMapping.java
@@ -0,0 +1,242 @@
+/*
+ *  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.gora.cassandra.store;
+
+import org.apache.gora.cassandra.bean.CassandraKey;
+import org.apache.gora.cassandra.bean.Field;
+import org.apache.gora.cassandra.bean.KeySpace;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * This class represents the Cassandra Mapping.
+ */
+public class CassandraMapping {
+
+  private static final String PRIMARY_KEY = "primarykey";
+  private CassandraKey cassandraKey;
+  private Map<String, String> tableProperties;
+  private Class keyClass;
+  private Class persistentClass;
+  private KeySpace keySpace;
+  private List<Field> fieldList;
+  private Field inlinedDefinedPartitionKey;
+  private String coreName;
+
+  /**
+   * Constructor of the class
+   */
+  CassandraMapping() {
+    this.fieldList = new ArrayList<>();
+    this.tableProperties = new HashMap<>();
+  }
+
+  /**
+   * This method returns the KeySpace in the mapping file,
+   *
+   * @return Key space {@link KeySpace}
+   */
+  public KeySpace getKeySpace() {
+    return keySpace;
+  }
+
+  /**
+   * This method set the KeySpace in the Cassandra mapping.
+   *
+   * @param keySpace Key space {@link KeySpace}
+   */
+  void setKeySpace(KeySpace keySpace) {
+    this.keySpace = keySpace;
+  }
+
+  /**
+   * Thi method returns only the fields which belongs only for the Persistent Object.
+   *
+   * @return List of Fields
+   */
+  public List<Field> getFieldList() {
+    return fieldList;
+  }
+
+  /**
+   * This method returns the Persistent Object's Field from the mapping, according to the FieldName.
+   *
+   * @param fieldName Field Name
+   * @return Field {@link Field}
+   */
+  public Field getFieldFromFieldName(String fieldName) {
+    for (Field field1 : fieldList) {
+      if (field1.getFieldName().equalsIgnoreCase(fieldName)) {
+        return field1;
+      }
+    }
+    return null;
+  }
+
+  /**
+   * This method returns the Persistent Object's Field from the mapping, according to the ColumnName.
+   *
+   * @param columnName Column Name
+   * @return Field {@link Field}
+   */
+  public Field getFieldFromColumnName(String columnName) {
+    for (Field field1 : fieldList) {
+      if (field1.getColumnName().equalsIgnoreCase(columnName)) {
+        return field1;
+      }
+    }
+    return null;
+  }
+
+  /**
+   * This method returns the Field Names
+   *
+   * @return array of Field Names
+   */
+  public String[] getFieldNames() {
+    List<String> fieldNames = new ArrayList<>(fieldList.size());
+    for (Field field : fieldList) {
+      fieldNames.add(field.getFieldName());
+    }
+    String[] fieldNameArray = new String[fieldNames.size()];
+    return fieldNames.toArray(fieldNameArray);
+  }
+
+  /**
+   * This method returns partition keys
+   *
+   * @return partitionKeys
+   */
+  public String[] getAllFieldsIncludingKeys() {
+    List<String> fieldNames = new ArrayList<>(fieldList.size());
+    for (Field field : fieldList) {
+      fieldNames.add(field.getFieldName());
+    }
+    if (cassandraKey != null) {
+      for (Field field : cassandraKey.getFieldList()) {
+        fieldNames.add(field.getFieldName());
+      }
+    }
+    String[] fieldNameArray = new String[fieldNames.size()];
+    return fieldNames.toArray(fieldNameArray);
+  }
+
+  /**
+   * This method return all the fields which involves with partition keys, Including composite Keys
+   * @return field Names
+   */
+  public String[] getAllKeys() {
+    List<String> fieldNames = new ArrayList<>();
+    Field keyField = getInlinedDefinedPartitionKey();
+    if (cassandraKey != null) {
+      for (Field field : cassandraKey.getFieldList()) {
+        fieldNames.add(field.getFieldName());
+      }
+    } else {
+      fieldNames.add(keyField.getFieldName());
+    }
+    String[] fieldNameArray = new String[fieldNames.size()];
+    return fieldNames.toArray(fieldNameArray);
+  }
+
+  public CassandraKey getCassandraKey() {
+    return cassandraKey;
+  }
+
+  void setCassandraKey(CassandraKey cassandraKey) {
+    this.cassandraKey = cassandraKey;
+  }
+
+  public String getCoreName() {
+    return coreName;
+  }
+
+  void setCoreName(String coreName) {
+    this.coreName = coreName;
+  }
+
+  void addCassandraField(Field field) {
+    this.fieldList.add(field);
+  }
+
+  void addProperty(String key, String value) {
+    this.tableProperties.put(key, value);
+  }
+
+  public String getProperty(String key) {
+    return this.tableProperties.get(key);
+  }
+
+  private Field getDefaultCassandraKey() {
+    Field field = new Field();
+    field.setFieldName("defaultId");
+    field.setColumnName("defaultId");
+    field.setType("varchar");
+    field.addProperty("primarykey", "true");
+    return field;
+  }
+
+  public Class getKeyClass() {
+    return keyClass;
+  }
+
+  public void setKeyClass(Class keyClass) {
+    this.keyClass = keyClass;
+  }
+
+  public Class getPersistentClass() {
+    return persistentClass;
+  }
+
+  void setPersistentClass(Class persistentClass) {
+    this.persistentClass = persistentClass;
+  }
+
+  /**
+   * This method return the Inlined defined Partition Key,
+   * If there isn't any inlined define partition keys,
+   * this method returns default predefined partition key "defaultId".
+   *
+   * @return Partition Key {@link Field}
+   */
+  public Field getInlinedDefinedPartitionKey() {
+    if (inlinedDefinedPartitionKey != null) {
+      return inlinedDefinedPartitionKey;
+    } else {
+      for (Field field : fieldList) {
+        if (Boolean.parseBoolean(field.getProperty(PRIMARY_KEY))) {
+          inlinedDefinedPartitionKey = field;
+          break;
+        }
+      }
+      if (inlinedDefinedPartitionKey == null) {
+        return getDefaultCassandraKey();
+      }
+      return inlinedDefinedPartitionKey;
+    }
+  }
+
+  void finalized() {
+    Field field = getInlinedDefinedPartitionKey();
+    if (!fieldList.contains(field) && cassandraKey == null) {
+      fieldList.add(field);
+    }
+  }
+}


Mime
View raw message