gora-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From lewi...@apache.org
Subject [2/3] gora git commit: Removing old gora-cassandra module
Date Wed, 23 Aug 2017 21:05:38 GMT
http://git-wip-us.apache.org/repos/asf/gora/blob/cbed41d0/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/TypeUtils.java
----------------------------------------------------------------------
diff --git a/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/TypeUtils.java b/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/TypeUtils.java
deleted file mode 100644
index c5db72b..0000000
--- a/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/TypeUtils.java
+++ /dev/null
@@ -1,232 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.gora.cassandra.serializers;
-
-import java.nio.ByteBuffer;
-import java.util.List;
-import java.util.Map;
-
-import org.apache.avro.Schema;
-import org.apache.avro.Schema.Type;
-import org.apache.avro.generic.GenericArray;
-import org.apache.avro.specific.SpecificFixed;
-import org.apache.avro.util.Utf8;
-import org.apache.gora.persistency.Persistent;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Utility class that infers the concrete Serializer needed to turn a value into
- * its binary representation
- */
-public class TypeUtils {
-
-  public static final Logger LOG = LoggerFactory.getLogger(TypeUtils.class);
-
-  // @SuppressWarnings({ "rawtypes", "unchecked" })
-  public static Class<? extends Object> getClass(Object value) {
-    return value.getClass();
-  }
-
-  public static Schema getSchema(Object value) {
-    if (value instanceof GenericArray) {
-      return Schema.createArray( getElementSchema((GenericArray<?>)value) );
-    } else {
-      return getSchema( getClass(value) );
-    }
-  }
-
-  public static Type getType(Object value) {
-    return getType( getClass(value) );
-  }
-
-  public static Type getType(Class<?> clazz) {
-    if (clazz.equals(Utf8.class)) {
-      return Type.STRING;
-    } else if (clazz.equals(Boolean.class) || clazz.equals(boolean.class)) {
-      return Type.BOOLEAN;
-    } else if (clazz.equals(ByteBuffer.class)) {
-      return Type.BYTES;
-    } else if (clazz.equals(Double.class) || clazz.equals(double.class)) {
-      return Type.DOUBLE;
-    } else if (clazz.equals(Float.class) || clazz.equals(float.class)) {
-      return Type.FLOAT;
-    } else if (clazz.equals(Integer.class) || clazz.equals(int.class)) {
-      return Type.INT;
-    } else if (clazz.equals(Long.class) || clazz.equals(long.class)) {
-      return Type.LONG;
-    } else if (clazz.isAssignableFrom(List.class)) {
-      return Type.ARRAY;
-    } else if (clazz.isAssignableFrom(Map.class)) {
-      return Type.MAP;
-    } else if (clazz.equals(Persistent.class)) {
-      return Type.RECORD;
-    } else if (clazz.getSuperclass().equals(SpecificFixed.class)) {
-      return Type.FIXED;
-    } else {
-      return null;
-    }
-  }
-
-  public static Class<?> getClass(Type type) {
-    if (type == Type.STRING) {
-      return Utf8.class;
-    } else if (type == Type.BOOLEAN) {
-      return Boolean.class;
-    } else if (type == Type.BYTES) {
-      return ByteBuffer.class;
-    } else if (type == Type.DOUBLE) {
-      return Double.class;
-    } else if (type == Type.FLOAT) {
-      return Float.class;
-    } else if (type == Type.INT) {
-      return Integer.class;
-    } else if (type == Type.LONG) {
-      return Long.class;
-    } else if (type == Type.ARRAY) {
-      return List.class;
-    } else if (type == Type.MAP) {
-      return Map.class;
-    } else if (type == Type.RECORD) {
-      return Persistent.class;
-    } else if (type == Type.FIXED) {
-      // return SpecificFixed.class;
-      return null;
-    } else {
-      return null;
-    }
-  }
-
-  public static Schema getSchema(Class<?> clazz) {
-    Type type = getType(clazz);
-    if (type == null) {
-      return null;
-    } else if (type == Type.FIXED) {
-      int size = getFixedSize(clazz);
-      String name = clazz.getName();
-      String space = null;
-      int n = name.lastIndexOf(".");
-      if (n < 0) {
-        space = name.substring(0,n);
-        name = name.substring(n+1);
-      } else {
-        space = null;
-      }
-      String doc = null; // ?
-      // LOG.info(Schema.createFixed(name, doc, space, size).toString());
-      return Schema.createFixed(name, doc, space, size);
-    } else if (type == Type.ARRAY) {
-      Object obj = null;
-      try {
-        obj = clazz.newInstance();
-      } catch (InstantiationException e) {
-        LOG.warn(e.toString());
-        return null;
-      } catch (IllegalAccessException e) {
-        LOG.warn(e.toString());
-        return null;
-      }
-      return getSchema(obj);
-    } else if (type == Type.MAP) {
-      // TODO
-      // return Schema.createMap(...);
-      return null;
-    } else if (type == Type.RECORD) {
-      // TODO
-      // return Schema.createRecord(...);
-      return null;
-    } else {
-      return Schema.create(type);
-    }
-  }
-
-  public static Class<?> getClass(Schema schema) {
-    Type type = schema.getType();
-    if (type == null) {
-      return null;
-    } else if (type == Type.FIXED) {
-      try {
-        return Class.forName( schema.getFullName() );
-      } catch (ClassNotFoundException e) {
-        LOG.warn(e.toString() + " : " + schema);
-        return null;
-      }
-    } else {
-      return getClass(type);
-    }
-  }
-
-  public static int getFixedSize(Type type) {
-    if (type == Type.BOOLEAN) {
-      return 1;
-    } else if (type == Type.DOUBLE) {
-      return 8;
-    } else if (type == Type.FLOAT) {
-      return 4;
-    } else if (type == Type.INT) {
-      return 4;
-    } else if (type == Type.LONG) {
-      return 8;
-    } else {
-      return -1;
-    }
-  }
-
-  public static int getFixedSize(Schema schema) {
-    Type type = schema.getType();
-    if (type == Type.FIXED) {
-      return schema.getFixedSize();
-    } else {
-      return getFixedSize(type);
-    }
-  }
-
-  public static int getFixedSize(Class<?> clazz) {
-    Type type = getType(clazz);
-    if (type == Type.FIXED) {
-      try {
-        return ((SpecificFixed)clazz.newInstance()).bytes().length;
-      } catch (InstantiationException e) {
-        LOG.warn(e.toString());
-        return -1;
-      } catch (IllegalAccessException e) {
-        LOG.warn(e.toString());
-        return -1;
-      }
-    } else {
-      return getFixedSize(type);
-    }
-  }
-
-  public static Schema getElementSchema(GenericArray<?> array) {
-    Schema schema = array.getSchema();
-    return (schema.getType() == Type.ARRAY) ? schema.getElementType() : schema;
-  }
-
-  /*
-  public static Schema getValueSchema(StatefulHashMap map) {
-    return map.getSchema().getValueType();
-  }
-
-  public static Type getValueType(StatefulHashMap map) {
-    return getValueSchema(map).getType();
-  }
-  */
-
-}

http://git-wip-us.apache.org/repos/asf/gora/blob/cbed41d0/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
deleted file mode 100644
index 5d22d94..0000000
--- a/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/package-info.java
+++ /dev/null
@@ -1,20 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-/**
- * 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/cbed41d0/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
deleted file mode 100644
index 1f9d614..0000000
--- a/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/CassandraClient.java
+++ /dev/null
@@ -1,658 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.gora.cassandra.store;
-
-import static org.apache.gora.cassandra.store.CassandraStore.colFamConsLvl;
-import static org.apache.gora.cassandra.store.CassandraStore.readOpConsLvl;
-import static org.apache.gora.cassandra.store.CassandraStore.writeOpConsLvl;
-
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-
-import me.prettyprint.cassandra.model.ConfigurableConsistencyLevel;
-import me.prettyprint.cassandra.serializers.ByteBufferSerializer;
-import me.prettyprint.cassandra.serializers.IntegerSerializer;
-import me.prettyprint.cassandra.serializers.StringSerializer;
-import me.prettyprint.cassandra.service.CassandraHostConfigurator;
-import me.prettyprint.hector.api.Cluster;
-import me.prettyprint.hector.api.Keyspace;
-import me.prettyprint.hector.api.beans.OrderedRows;
-import me.prettyprint.hector.api.beans.OrderedSuperRows;
-import me.prettyprint.hector.api.beans.Row;
-import me.prettyprint.hector.api.beans.SuperRow;
-import me.prettyprint.hector.api.ddl.ColumnFamilyDefinition;
-import me.prettyprint.hector.api.ddl.ComparatorType;
-import me.prettyprint.hector.api.ddl.KeyspaceDefinition;
-import me.prettyprint.hector.api.factory.HFactory;
-import me.prettyprint.hector.api.mutation.Mutator;
-import me.prettyprint.hector.api.query.QueryResult;
-import me.prettyprint.hector.api.query.RangeSlicesQuery;
-import me.prettyprint.hector.api.query.RangeSuperSlicesQuery;
-import me.prettyprint.hector.api.HConsistencyLevel;
-import me.prettyprint.hector.api.Serializer;
-
-import org.apache.avro.Schema;
-import org.apache.avro.Schema.Type;
-import org.apache.avro.generic.GenericArray;
-import org.apache.gora.cassandra.query.CassandraQuery;
-import org.apache.gora.cassandra.serializers.GoraSerializerTypeInferer;
-import org.apache.gora.mapreduce.GoraRecordReader;
-import org.apache.gora.persistency.impl.PersistentBase;
-import org.apache.gora.query.Query;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * CassandraClient is where all of the primary datastore functionality is 
- * executed. Typically CassandraClient is invoked by calling 
- * {@link org.apache.gora.cassandra.store.CassandraStore#initialize(Class, Class, Properties)}.
- * CassandraClient deals with Cassandra data model definition, mutation, 
- * and general/specific mappings.
- * {@link org.apache.gora.cassandra.store.CassandraStore#initialize} .
- *
- * @param <K>
- * @param <T>
- */
-public class CassandraClient<K, T extends PersistentBase> {
-  
-  /** The logging implementation */
-  public static final Logger LOG = LoggerFactory.getLogger(CassandraClient.class);
-  
-  private Cluster cluster;
-  private Keyspace keyspace;
-  private Mutator<K> mutator;
-  private Class<K> keyClass;
-  private Class<T> persistentClass;
-  
-  /** Object which holds the XML mapping for Cassandra. */
-  private CassandraMapping cassandraMapping = null;
-
-  /** Hector client default column family consistency level. */
-  public static final String DEFAULT_HECTOR_CONSIS_LEVEL = "QUORUM";
-  
-  /** Cassandra serializer to be used for serializing Gora's keys. */
-  private Serializer<K> keySerializer;
-
-  /**
-   * Method to maintain backward compatibility with earlier versions.
-   * @param keyClass
-   * @param persistentClass
-   * @throws Exception
-   */
-  public void initialize(Class<K> keyClass, Class<T> persistentClass)
-    throws Exception {
-	initialize(keyClass, persistentClass, null);
-  }
-
-  /**
-   * Given our key, persistentClass from
-   * {@link org.apache.gora.cassandra.store.CassandraStore#initialize(Class, Class, Properties)}
-   * we make best efforts to dictate our data model.
-   * We make a quick check within {@link org.apache.gora.cassandra.store.CassandraClient#checkKeyspace() }
-   * to see if our keyspace has already been invented, this simple check prevents us from
-   * recreating the keyspace if it already exists.
-   * We then simple specify (based on the input keyclass) an appropriate serializer
-   * via {@link org.apache.gora.cassandra.serializers.GoraSerializerTypeInferer} before
-   * defining a mutator from and by which we can mutate this object.
-   *
-   * @param keyClass        the Key by which we wish o assign a record object
-   * @param persistentClass the generated {@link org.apache.gora.persistency.Persistent} bean representing the data.
-   * @param properties      key value pairs from gora.properties
-   * @throws Exception
-   */
-  public void initialize(Class<K> keyClass, Class<T> persistentClass, Properties properties) throws Exception {
-    this.keyClass = keyClass;
-
-    // get cassandra mapping with persistent class
-    this.persistentClass = persistentClass;
-    this.cassandraMapping = CassandraMappingManager.getManager().get(persistentClass);
-    Map<String, String> accessMap = null;
-    if (properties != null) {
-      String username = properties
-          .getProperty("gora.cassandrastore.username");
-      if (username != null) {
-        accessMap = new HashMap<>();
-        accessMap.put("username", username);
-        String password = properties
-            .getProperty("gora.cassandrastore.password");
-        if (password != null) {
-          accessMap.put("password", password);
-        }
-      }
-    }
-
-    this.cluster = HFactory.getOrCreateCluster(this.cassandraMapping.getClusterName(), 
-        new CassandraHostConfigurator(this.cassandraMapping.getHostName()), accessMap);
-    
-    // add keyspace to cluster
-    checkKeyspace();
-    
-    // Just create a Keyspace object on the client side, corresponding to an already 
-    // existing keyspace with already created column families.
-    this.keyspace = HFactory.createKeyspace(this.cassandraMapping.getKeyspaceName(), this.cluster);
-    
-    this.keySerializer = GoraSerializerTypeInferer.getSerializer(keyClass);
-    if (this.keySerializer == null)
-      LOG.error("Serializer for " + keyClass + " not found.");
-    this.mutator = HFactory.createMutator(this.keyspace, this.keySerializer);
-  }
-
-  /**
-   * Check if keyspace already exists.
-   *
-   * @return if keyspace already exists return true.
-   */
-  public boolean keyspaceExists() {
-    KeyspaceDefinition keyspaceDefinition = this.cluster.describeKeyspace(this.cassandraMapping.getKeyspaceName());
-    return (keyspaceDefinition != null);
-  }
-
-  /**
-   * Check if keyspace already exists. If not, create it.
-   * In this method, we also utilize Hector's
-   * {@link me.prettyprint.cassandra.model.ConfigurableConsistencyLevel} logic.
-   * It is set by passing a
-   * {@link me.prettyprint.cassandra.model.ConfigurableConsistencyLevel} object right
-   * when the {@link me.prettyprint.hector.api.Keyspace} is created.
-   * If we cannot find a consistency level within <code>gora.properites</code>,
-   * then column family consistency level is set to QUORUM (by default) which permits
-   * consistency to wait for a quorum of replicas to respond regardless of data center.
-   * QUORUM is Hector Client's default setting and we respect that here as well.
-   *
-   * @see <a href="http://hector-client.github.io/hector/build/html/content/consistency_level.html">Consistency Level</a>
-   */
-  public void checkKeyspace() {
-    // "describe keyspace <keyspaceName>;" query
-    KeyspaceDefinition keyspaceDefinition = this.cluster.describeKeyspace(this.cassandraMapping.getKeyspaceName());
-    if (keyspaceDefinition == null) {
-      List<ColumnFamilyDefinition> columnFamilyDefinitions = this.cassandraMapping.getColumnFamilyDefinitions();
-
-      // GORA-197
-      for (ColumnFamilyDefinition cfDef : columnFamilyDefinitions) {
-        cfDef.setComparatorType(ComparatorType.BYTESTYPE);
-      }
-
-      keyspaceDefinition = HFactory.createKeyspaceDefinition(
-        this.cassandraMapping.getKeyspaceName(), 
-        this.cassandraMapping.getKeyspaceReplicationStrategy(),
-        this.cassandraMapping.getKeyspaceReplicationFactor(),
-        columnFamilyDefinitions
-      );
-      
-      this.cluster.addKeyspace(keyspaceDefinition, true);
-      
-      // GORA-167 Create a customized Consistency Level
-      ConfigurableConsistencyLevel ccl = new ConfigurableConsistencyLevel();
-      Map<String, HConsistencyLevel> clmap = getConsisLevelForColFams(columnFamilyDefinitions);
-      // Column family consistency levels
-      ccl.setReadCfConsistencyLevels(clmap);
-      ccl.setWriteCfConsistencyLevels(clmap);
-      // Operations consistency levels
-      String opConsisLvl = (readOpConsLvl!=null && !readOpConsLvl.isEmpty())?readOpConsLvl:DEFAULT_HECTOR_CONSIS_LEVEL;
-      ccl.setDefaultReadConsistencyLevel(HConsistencyLevel.valueOf(opConsisLvl));
-      LOG.debug("Hector read consistency configured to '" + opConsisLvl + "'.");
-      opConsisLvl = (writeOpConsLvl!=null && !writeOpConsLvl.isEmpty())?writeOpConsLvl:DEFAULT_HECTOR_CONSIS_LEVEL;
-      ccl.setDefaultWriteConsistencyLevel(HConsistencyLevel.valueOf(opConsisLvl));
-      LOG.debug("Hector write consistency configured to '" + opConsisLvl + "'.");
-
-      HFactory.createKeyspace("Keyspace", this.cluster, ccl);
-      keyspaceDefinition = null;
-    }
-    else {
-      List<ColumnFamilyDefinition> cfDefs = keyspaceDefinition.getCfDefs();
-      if (cfDefs == null || cfDefs.size() == 0) {
-        LOG.warn(keyspaceDefinition.getName() + " does not have any column family.");
-      }
-      else {
-        for (ColumnFamilyDefinition cfDef : cfDefs) {
-          ComparatorType comparatorType = cfDef.getComparatorType();
-          if (! comparatorType.equals(ComparatorType.BYTESTYPE)) {
-            // GORA-197
-            LOG.warn("The comparator type of " + cfDef.getName() + " column family is " + comparatorType.getTypeName()
-              + ", not BytesType. It may cause a fatal error on column validation later.");
-          }
-          else {
-            LOG.debug("The comparator type of " + cfDef.getName() + " column family is " 
-              + comparatorType.getTypeName() + ".");
-          }
-        }
-      }
-    }
-  }
-
-  /**
-   * Method in charge of setting the consistency level for defined column families.
-   * @param pColFams  Column families
-   * @return Map<String, HConsistencyLevel> with the mapping between colFams and consistency level.
-   */
-  private Map<String, HConsistencyLevel> getConsisLevelForColFams(List<ColumnFamilyDefinition> pColFams) {
-    Map<String, HConsistencyLevel> clMap = new HashMap<>();
-    // Get columnFamily consistency level.
-    String colFamConsisLvl = (colFamConsLvl != null && !colFamConsLvl.isEmpty())?colFamConsLvl:DEFAULT_HECTOR_CONSIS_LEVEL;
-    LOG.debug("ColumnFamily consistency level configured to '" + colFamConsisLvl + "'.");
-    // Define consistency for ColumnFamily "ColumnFamily"
-    for (ColumnFamilyDefinition colFamDef : pColFams)
-      clMap.put(colFamDef.getName(), HConsistencyLevel.valueOf(colFamConsisLvl));
-    return clMap;
-  }
-  
-  /**
-   * Drop keyspace.
-   */
-  public void dropKeyspace() {
-    this.cluster.dropKeyspace(this.cassandraMapping.getKeyspaceName());
-  }
-
-  /**
-   * Insert a field in a column.
-   * @param key the row key
-   * @param fieldName the field name
-   * @param value the field value.
-   */
-  public void addColumn(K key, String fieldName, Object value) {
-    if (value == null) {
-    	LOG.debug( "field:"+fieldName+", its value is null.");
-      return;
-    }
-
-    ByteBuffer byteBuffer = toByteBuffer(value);
-    String columnFamily = this.cassandraMapping.getFamily(fieldName);
-    String columnName = this.cassandraMapping.getColumn(fieldName);
-    
-    if (columnName == null) {
-    	LOG.warn("Column name is null for field: " + fieldName );
-        return;
-    }
-      
-    if( LOG.isDebugEnabled() ) LOG.debug( "fieldName: "+fieldName +" columnName: " + columnName );
-    
-    String ttlAttr = this.cassandraMapping.getColumnsAttribs().get(columnName);
-    
-    if ( null == ttlAttr ){
-    	ttlAttr = CassandraMapping.DEFAULT_COLUMNS_TTL;
-    	if( LOG.isDebugEnabled() ) LOG.debug( "ttl was not set for field: " + fieldName + ". Using " + ttlAttr );
-    } else {
-    	if( LOG.isDebugEnabled() ) LOG.debug( "ttl for field: " + fieldName + " is " + ttlAttr );
-    }
-
-    synchronized(mutator) {
-      HectorUtils.insertColumn(mutator, key, columnFamily, columnName, byteBuffer, ttlAttr);
-    }
-  }
-
-  /**
-   * Delete a row within the keyspace.
-   *
-   * @param key
-   * @param familyName
-   * @param columnName
-   */
-  public void deleteColumn(K key, String familyName, ByteBuffer columnName) {
-    synchronized(mutator) {
-      HectorUtils.deleteColumn(mutator, key, familyName, columnName);
-    }
-  }
-
-  /**
-   * Deletes an entry based on its key.
-   * @param key
-   */
-  public void deleteByKey(K key) {
-    Map<String, String> familyMap = this.cassandraMapping.getFamilyMap();
-    deleteColumn(key, familyMap.values().iterator().next(), null);
-  }
-
-  /**
-   * Insert a member in a super column. This is used for map and record Avro types.
-   * @param key the row key
-   * @param fieldName the field name
-   * @param columnName the column name (the member name, or the index of array)
-   * @param value the member value
-   */
-  public void addSubColumn(K key, String fieldName, ByteBuffer columnName, Object value) {
-    if (value == null) {
-      return;
-    }
-
-    ByteBuffer byteBuffer = toByteBuffer(value);
-    
-    String columnFamily = this.cassandraMapping.getFamily(fieldName);
-    String superColumnName = this.cassandraMapping.getColumn(fieldName);
-    String ttlAttr = this.cassandraMapping.getColumnsAttribs().get(superColumnName);
-    if ( null == ttlAttr ) {
-      ttlAttr = CassandraMapping.DEFAULT_COLUMNS_TTL;
-      if( LOG.isDebugEnabled() ) LOG.debug( "ttl was not set for field:" + fieldName + " .Using " + ttlAttr );
-    } else {
-      if( LOG.isDebugEnabled() ) LOG.debug( "ttl for field:" + fieldName + " is " + ttlAttr );
-    }
-
-    synchronized(mutator) {
-      HectorUtils.insertSubColumn(mutator, key, columnFamily, superColumnName, columnName, byteBuffer, ttlAttr);
-    }
-  }
-
-  /**
-   * Adds an subColumn inside the cassandraMapping file when a String is serialized
-   * @param key the row key
-   * @param fieldName the field name
-   * @param columnName the column name (the member name, or the index of array)
-   * @param value the member value
-   */
-  public void addSubColumn(K key, String fieldName, String columnName, Object value) {
-    addSubColumn(key, fieldName, StringSerializer.get().toByteBuffer(columnName), value);
-  }
-
-  /**
-   * Adds an subColumn inside the cassandraMapping file when an Integer is serialized
-   * @param key the row key
-   * @param fieldName the field name
-   * @param columnName the column name (the member name, or the index of array)
-   * @param value the member value
-   */
-  public void addSubColumn(K key, String fieldName, Integer columnName, Object value) {
-    addSubColumn(key, fieldName, IntegerSerializer.get().toByteBuffer(columnName), value);
-  }
-
-
-  /**
-   * Delete a member in a super column. This is used for map and record Avro types.
-   * @param key the row key
-   * @param fieldName the field name
-   * @param columnName the column name (the member name, or the index of array)
-   */
-  public void deleteSubColumn(K key, String fieldName, ByteBuffer columnName) {
-
-    String columnFamily = this.cassandraMapping.getFamily(fieldName);
-    String superColumnName = this.cassandraMapping.getColumn(fieldName);
-    
-    synchronized(mutator) {
-      HectorUtils.deleteSubColumn(mutator, key, columnFamily, superColumnName, columnName);
-    }
-  }
-
-  /**
-   * Deletes a subColumn 
-   * @param key
-   * @param fieldName
-   * @param columnName
-   */
-  public void deleteSubColumn(K key, String fieldName, String columnName) {
-    deleteSubColumn(key, fieldName, StringSerializer.get().toByteBuffer(columnName));
-  }
-
-  /**
-   * Deletes all subcolumns from a super column.
-   * @param key the row key.
-   * @param fieldName the field name.
-   */
-  public void deleteSubColumn(K key, String fieldName) {
-    String columnFamily = this.cassandraMapping.getFamily(fieldName);
-    String superColumnName = this.cassandraMapping.getColumn(fieldName);
-    synchronized(mutator) {
-      HectorUtils.deleteSubColumn(mutator, key, columnFamily, superColumnName, null);
-    }
-  }
-
-  public void deleteGenericArray(K key, String fieldName) {
-    //TODO Verify this. Everything that goes inside a genericArray will go inside a column so let's just delete that.
-    deleteColumn(key, cassandraMapping.getFamily(fieldName), toByteBuffer(fieldName));
-  }
-  
-  public void addGenericArray(K key, String fieldName, GenericArray<?> array) {
-    if (isSuper( cassandraMapping.getFamily(fieldName) )) {
-      int i= 0;
-      for (Object itemValue: array) {
-
-        // TODO: hack, do not store empty arrays
-        if (itemValue instanceof GenericArray<?>) {
-          if (((List<?>)itemValue).size() == 0) {
-            continue;
-          }
-        } else if (itemValue instanceof Map<?,?>) {
-          if (((Map<?, ?>)itemValue).size() == 0) {
-            continue;
-          }
-        }
-
-        addSubColumn(key, fieldName, i++, itemValue);
-      }
-    }
-    else {
-      addColumn(key, fieldName, array);
-    }
-  }
-
-  public void deleteStatefulHashMap(K key, String fieldName) {
-    if (isSuper( cassandraMapping.getFamily(fieldName) )) {
-      deleteSubColumn(key, fieldName);
-    } else {
-      deleteColumn(key, cassandraMapping.getFamily(fieldName), toByteBuffer(fieldName));
-    }
-  }
-
-  public void addStatefulHashMap(K key, String fieldName, Map<CharSequence,Object> map) {
-    if (isSuper( cassandraMapping.getFamily(fieldName) )) {
-      // as we don't know what has changed inside the map or If it's an empty map, then delete its content.
-      deleteSubColumn(key, fieldName);
-      // update if there is anything to update.
-      if (!map.isEmpty()) {
-        // If it's not empty, then update its content.
-        for (CharSequence mapKey: map.keySet()) {
-          // TODO: hack, do not store empty arrays
-          Object mapValue = map.get(mapKey);
-          if (mapValue instanceof GenericArray<?>) {
-            if (((List<?>)mapValue).size() == 0) {
-              continue;
-            }
-          } else if (mapValue instanceof Map<?,?>) {
-            if (((Map<?, ?>)mapValue).size() == 0) {
-              continue;
-            }
-          }
-          addSubColumn(key, fieldName, mapKey.toString(), mapValue);
-        }
-      }
-    }
-    else {
-      addColumn(key, fieldName, map);
-    }
-  }
-
-  /**
-   * Serialize value to ByteBuffer using 
-   * {@link org.apache.gora.cassandra.serializers.GoraSerializerTypeInferer#getSerializer(Object)}.
-   * @param value the member value {@link java.lang.Object}.
-   * @return ByteBuffer object
-   */
-  public ByteBuffer toByteBuffer(Object value) {
-    ByteBuffer byteBuffer = null;
-    Serializer<Object> serializer = GoraSerializerTypeInferer.getSerializer(value);
-    if (serializer == null) {
-      LOG.warn("Serializer not found for: " + value.toString());
-    }
-    else {
-      LOG.debug(serializer.getClass() + " selected as appropriate Serializer.");
-      byteBuffer = serializer.toByteBuffer(value);
-    }
-    if (byteBuffer == null) {
-      LOG.warn("Serialization value for: " + value.getClass().getName() + " = null");
-    }
-    return byteBuffer;
-  }
-
-  /**
-   * Select a family column in the keyspace.
-   * @param cassandraQuery a wrapper of the query
-   * @param family the family name to be queried
-   * @return a list of family rows
-   */
-  public List<Row<K, ByteBuffer, ByteBuffer>> execute(CassandraQuery<K, T> cassandraQuery, String family) {
-    
-    String[] columnNames = cassandraQuery.getColumns(family);
-    ByteBuffer[] columnNameByteBuffers = new ByteBuffer[columnNames.length];
-    for (int i = 0; i < columnNames.length; i++) {
-      columnNameByteBuffers[i] = StringSerializer.get().toByteBuffer(columnNames[i]);
-    }
-    Query<K, T> query = cassandraQuery.getQuery();
-    int limit = (int) query.getLimit();
-    if (limit < 1) {
-      limit = Integer.MAX_VALUE;
-    }
-    K startKey = query.getStartKey();
-    K endKey = query.getEndKey();
-    
-    RangeSlicesQuery<K, ByteBuffer, ByteBuffer> rangeSlicesQuery = HFactory.createRangeSlicesQuery
-        (this.keyspace, this.keySerializer, ByteBufferSerializer.get(), ByteBufferSerializer.get());
-    rangeSlicesQuery.setColumnFamily(family);
-    rangeSlicesQuery.setKeys(startKey, endKey);
-    rangeSlicesQuery.setRange(ByteBuffer.wrap(new byte[0]), ByteBuffer.wrap(new byte[0]), false, GoraRecordReader.BUFFER_LIMIT_READ_VALUE);
-    rangeSlicesQuery.setRowCount(limit);
-    rangeSlicesQuery.setColumnNames(columnNameByteBuffers);
-    
-    QueryResult<OrderedRows<K, ByteBuffer, ByteBuffer>> queryResult = rangeSlicesQuery.execute();
-    OrderedRows<K, ByteBuffer, ByteBuffer> orderedRows = queryResult.get();
-    
-    return orderedRows.getList();
-  }
-  
-  private String getMappingFamily(String pField){
-    String family = null;
-    // checking if it was a UNION field the one we are retrieving
-    if (pField.indexOf(CassandraStore.UNION_COL_SUFIX) > 0)
-      family = this.cassandraMapping.getFamily(pField.substring(0,pField.indexOf(CassandraStore.UNION_COL_SUFIX)));
-    else
-      family = this.cassandraMapping.getFamily(pField);
-     return family;
-   }
- 
-  private String getMappingColumn(String pField){
-    String column = null;
-    if (pField.indexOf(CassandraStore.UNION_COL_SUFIX) > 0)
-      column = pField;
-    else
-      column = this.cassandraMapping.getColumn(pField);
-      return column;
-    }
-
-  /**
-   * Select the families that contain at least one column mapped to a query field.
-   * @param query indicates the columns to select
-   * @return a map which keys are the family names and values the 
-   * corresponding column names required to get all the query fields.
-   */
-  public Map<String, List<String>> getFamilyMap(Query<K, T> query) {
-    Map<String, List<String>> map = new HashMap<>();
-    Schema persistentSchema = query.getDataStore().newPersistent().getSchema();
-    for (String field: query.getFields()) {
-      String family = this.getMappingFamily(field);
-      String column = this.getMappingColumn(field);
-      
-      // check if the family value was already initialized 
-      List<String> list = map.get(family);
-      if (list == null) {
-        list = new ArrayList<>();
-        map.put(family, list);
-      }
-      if (persistentSchema.getField(field).schema().getType() == Type.UNION)
-        list.add(field + CassandraStore.UNION_COL_SUFIX);
-      if (column != null) {
-        list.add(column);
-      }
-    }
-    
-    return map;
-  }
-
-  /**
-   * Retrieves the cassandraMapping which holds whatever was mapped 
-   * from the gora-cassandra-mapping.xml
-   * @return 
-   */
-  public CassandraMapping getCassandraMapping(){
-    return this.cassandraMapping;
-  }
-  
-  /**
-   * Select the field names according to the column names, which format 
-   * if fully qualified: "family:column"
-   * @param query
-   * @return a map which keys are the fully qualified column 
-   * names and values the query fields
-   */
-  public Map<String, String> getReverseMap(Query<K, T> query) {
-    Map<String, String> map = new HashMap<>();
-    Schema persistentSchema = query.getDataStore().newPersistent().getSchema();
-    for (String field: query.getFields()) {
-      String family = this.getMappingFamily(field);
-      String column = this.getMappingColumn(field);
-      if (persistentSchema.getField(field).schema().getType() == Type.UNION)
-        map.put(family + ":" + field + CassandraStore.UNION_COL_SUFIX, field + CassandraStore.UNION_COL_SUFIX);
-      map.put(family + ":" + column, field);
-    }
-    
-    return map;
-  }
-
-  /**
-   * Determines if a column is a superColumn or not.
-   * @param family
-   * @return boolean
-   */
-  public boolean isSuper(String family) {
-    return this.cassandraMapping.isSuper(family);
-  }
-
-  public List<SuperRow<K, String, ByteBuffer, ByteBuffer>> executeSuper(CassandraQuery<K, T> cassandraQuery, String family) {
-    String[] columnNames = cassandraQuery.getColumns(family);
-    Query<K, T> query = cassandraQuery.getQuery();
-    int limit = (int) query.getLimit();
-    if (limit < 1) {
-      limit = Integer.MAX_VALUE;
-    }
-    K startKey = query.getStartKey();
-    K endKey = query.getEndKey();
-    
-    RangeSuperSlicesQuery<K, String, ByteBuffer, ByteBuffer> rangeSuperSlicesQuery = HFactory.createRangeSuperSlicesQuery
-        (this.keyspace, this.keySerializer, StringSerializer.get(), ByteBufferSerializer.get(), ByteBufferSerializer.get());
-    rangeSuperSlicesQuery.setColumnFamily(family);    
-    rangeSuperSlicesQuery.setKeys(startKey, endKey);
-    rangeSuperSlicesQuery.setRange("", "", false, GoraRecordReader.BUFFER_LIMIT_READ_VALUE);
-    rangeSuperSlicesQuery.setRowCount(limit);
-    rangeSuperSlicesQuery.setColumnNames(columnNames);
-    
-    
-    QueryResult<OrderedSuperRows<K, String, ByteBuffer, ByteBuffer>> queryResult = rangeSuperSlicesQuery.execute();
-    OrderedSuperRows<K, String, ByteBuffer, ByteBuffer> orderedRows = queryResult.get();
-    return orderedRows.getList();
-
-
-  }
-
-  /**
-   * Obtain Schema/Keyspace name
-   * @return Keyspace
-   */
-  public String getKeyspaceName() {
-    return this.cassandraMapping.getKeyspaceName();
-  }
-}

http://git-wip-us.apache.org/repos/asf/gora/blob/cbed41d0/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
deleted file mode 100644
index b7a7087..0000000
--- a/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/CassandraMapping.java
+++ /dev/null
@@ -1,329 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.gora.cassandra.store;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-import me.prettyprint.cassandra.model.BasicColumnFamilyDefinition;
-import me.prettyprint.cassandra.service.ThriftCfDef;
-import me.prettyprint.hector.api.ddl.ColumnFamilyDefinition;
-import me.prettyprint.hector.api.ddl.ColumnType;
-import me.prettyprint.hector.api.ddl.ComparatorType;
-
-import org.jdom.Element;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Mapping definitions for CouchDB.
- */
-public class CassandraMapping {
-  
-  public static final Logger LOG = LoggerFactory.getLogger(CassandraMapping.class);
-  
-  private static final String NAME_ATTRIBUTE = "name";
-  private static final String COLUMN_ATTRIBUTE = "qualifier";
-  private static final String FAMILY_ATTRIBUTE = "family";
-  private static final String SUPER_ATTRIBUTE = "type";
-  private static final String CLUSTER_ATTRIBUTE = "cluster";
-  private static final String HOST_ATTRIBUTE = "host";
-
-  private static final String GCGRACE_SECONDS_ATTRIBUTE = "gc_grace_seconds";
-  private static final String COLUMNS_TTL_ATTRIBUTE = "ttl";
-  private static final String REPLICATION_FACTOR_ATTRIBUTE = "replication_factor"; 	
-  private static final String REPLICATION_STRATEGY_ATTRIBUTE = "placement_strategy";
-  
-  public static final String DEFAULT_REPLICATION_FACTOR = "1";		 
-  public static final String DEFAULT_REPLICATION_STRATEGY = "org.apache.cassandra.locator.SimpleStrategy";
-  public static final String DEFAULT_COLUMNS_TTL = "0";
-  public static final String DEFAULT_GCGRACE_SECONDS = "0";
-
-  private String hostName;
-  private String clusterName;
-  private String keyspaceName;
-  private String keyspaceStrategy;
-  private int 	 keyspaceRF;
-  
-  
-  /**
-   * List of the super column families.
-   */
-  private List<String> superFamilies = new ArrayList<>();
-
-  /**
-   * Look up the column family associated to the Avro field.
-   */
-  private Map<String, String> familyMap = new HashMap<>();
-  
-  /**
-   * Look up the column associated to the Avro field.
-   */
-  private Map<String, String> columnMap = new HashMap<>();
-
-  /**
-   * Helps storing attributes defined for each field.
-   */
-  private Map<String, String> columnAttrMap = new HashMap<>();
-  
-  /**
-   * Look up the column family from its name.
-   */
-  private Map<String, BasicColumnFamilyDefinition> columnFamilyDefinitions = 
-		  new HashMap<>();
-
-  
-  /**
-   * Simply gets the Cassandra host name.
-   * @return hostName
-   */
-  public String getHostName() {
-    return this.hostName;
-  }
-  
-  /**
-   * Simply gets the Cassandra cluster (the machines (nodes) 
-   * in a logical Cassandra instance) name.
-   * Clusters can contain multiple keyspaces. 
-   * @return clusterName
-   */
-  public String getClusterName() {
-    return this.clusterName;
-  }
-
-  /**
-   * Simply gets the Cassandra namespace for ColumnFamilies, typically one per application
-   * @return
-   */
-  public String getKeyspaceName() {
-    return this.keyspaceName;
-  }
-  
-  /**
-   * gets the replication strategy
-   * @return string class name to be used for strategy
-   */
-  public String getKeyspaceReplicationStrategy() {
-	return this.keyspaceStrategy;
-  }
-  
-  /**
-   * gets the replication factor
-   * @return int replication factor
-   */
-  public int getKeyspaceReplicationFactor() {
-	return this.keyspaceRF;
-  }
-
-  /**
-   * Primary class for loading Cassandra configuration from the 'MAPPING_FILE'.
-   * It should be noted that should the "qualifier" attribute and its associated
-   * value be absent from class field definition, it will automatically be set to 
-   * the field name value.
-   * 
-   */
-  @SuppressWarnings("unchecked")
-  public CassandraMapping(Element keyspace, Element mapping) {
-    if (keyspace == null) {
-      LOG.error("Keyspace element should not be null!");
-      return;
-    } else {
-      if (LOG.isDebugEnabled()) {
-        LOG.debug("Located Cassandra Keyspace");
-      }
-    }
-    this.keyspaceName = keyspace.getAttributeValue(NAME_ATTRIBUTE);
-    if (this.keyspaceName == null) {
-      LOG.error("Error locating Cassandra Keyspace name attribute!");
-    } else {
-      if (LOG.isDebugEnabled()) {
-        LOG.debug("Located Cassandra Keyspace name: '" + keyspaceName + "'");
-      }
-    }
-    this.clusterName = keyspace.getAttributeValue(CLUSTER_ATTRIBUTE);
-    if (this.clusterName == null) {
-    	LOG.error("Error locating Cassandra Keyspace cluster attribute!");
-    } else {
-      if (LOG.isDebugEnabled()) {
-        LOG.debug("Located Cassandra Keyspace cluster: '" + clusterName + "'");
-      }
-    }
-    this.hostName = keyspace.getAttributeValue(HOST_ATTRIBUTE);
-    if (this.hostName == null) {
-    	LOG.error("Error locating Cassandra Keyspace host attribute!");
-    } else {
-      if (LOG.isDebugEnabled()) {
-        LOG.debug("Located Cassandra Keyspace host: '" + hostName + "'");
-      }  
-    }
-    
-    // setting replication strategy
-    this.keyspaceStrategy = keyspace.getAttributeValue( REPLICATION_STRATEGY_ATTRIBUTE );
-    if( null == this.keyspaceStrategy ) {
-    	this.keyspaceStrategy = DEFAULT_REPLICATION_STRATEGY;
-    }
-	if( LOG.isDebugEnabled() ) {
-		LOG.debug( "setting Keyspace replication strategy to " + this.keyspaceStrategy );
-	}
-
-	// setting replication factor
-	String tmp = keyspace.getAttributeValue( REPLICATION_FACTOR_ATTRIBUTE );	
-	if( null == tmp ) {
-		tmp = DEFAULT_REPLICATION_FACTOR;
-	}
-	this.keyspaceRF = Integer.parseInt( tmp );
-	if( LOG.isDebugEnabled() ) {
-		LOG.debug( "setting Keyspace replication factor to " + this.keyspaceRF );
-	}
-
-    
-    // load column family definitions
-    List<Element> elements = keyspace.getChildren();
-    for (Element element: elements) {
-      BasicColumnFamilyDefinition cfDef = new BasicColumnFamilyDefinition();
-      
-      String familyName = element.getAttributeValue(NAME_ATTRIBUTE);
-      if (familyName == null) {
-      	LOG.error("Error locating column family name attribute!");
-      	continue;
-      } else {
-        if (LOG.isDebugEnabled()) {
-          LOG.debug("Located column family: '" + familyName + "'" );
-        }
-      }
-      String gcgrace_scs = element.getAttributeValue(GCGRACE_SECONDS_ATTRIBUTE);
-      if (gcgrace_scs == null) {
-        LOG.warn("Error locating gc_grace_seconds attribute for '" + familyName + "' column family");
-        LOG.warn("Using gc_grace_seconds default value which is: " + DEFAULT_GCGRACE_SECONDS 
-        		+ " and is viable ONLY FOR A SINGLE NODE CLUSTER");
-        LOG.warn("please update the gora-cassandra-mapping.xml file to avoid seeing this warning");
-      } else {
-        if (LOG.isDebugEnabled()) {
-        LOG.debug("Located gc_grace_seconds: '" + gcgrace_scs + "'" );
-        }
-      }
-
-      String superAttribute = element.getAttributeValue(SUPER_ATTRIBUTE);
-      if (superAttribute != null) {
-        if (LOG.isDebugEnabled()) {
-          LOG.debug("Located super column family");
-        }
-        this.superFamilies.add(familyName);
-        if (LOG.isDebugEnabled()) {
-          LOG.debug("Added super column family: '" + familyName + "'");
-        }
-        cfDef.setColumnType(ColumnType.SUPER);
-        cfDef.setSubComparatorType(ComparatorType.BYTESTYPE);
-      }
-
-      cfDef.setKeyspaceName(this.keyspaceName);
-      cfDef.setName(familyName);
-      cfDef.setComparatorType(ComparatorType.BYTESTYPE);
-      cfDef.setDefaultValidationClass(ComparatorType.BYTESTYPE.getClassName());
-
-      cfDef.setGcGraceSeconds(Integer.parseInt( gcgrace_scs!=null?gcgrace_scs:DEFAULT_GCGRACE_SECONDS));
-      this.columnFamilyDefinitions.put(familyName, cfDef);
-
-    }
-    
-    // load column definitions    
-    elements = mapping.getChildren();
-    for (Element element: elements) {
-      String fieldName = element.getAttributeValue(NAME_ATTRIBUTE);
-      String familyName = element.getAttributeValue(FAMILY_ATTRIBUTE);
-      String columnName = element.getAttributeValue(COLUMN_ATTRIBUTE);
-      String ttlValue = element.getAttributeValue(COLUMNS_TTL_ATTRIBUTE);
-      if (fieldName == null) {
-       LOG.error("Field name is not declared.");
-        continue;
-      }
-      if (familyName == null) {
-        LOG.error("Family name is not declared for \"" + fieldName + "\" field.");
-        continue;
-      }
-      if (columnName == null) {
-        LOG.warn("Column name (qualifier) is not declared for \"" + fieldName + "\" field.");
-        columnName = fieldName;
-      }
-      if (ttlValue == null) {
-        LOG.warn("TTL value is not defined for \"" + fieldName + "\" field. \n Using default value: " + DEFAULT_COLUMNS_TTL);
-      }
-
-      BasicColumnFamilyDefinition columnFamilyDefinition = this.columnFamilyDefinitions.get(familyName);
-      if (columnFamilyDefinition == null) {
-        LOG.warn("Family " + familyName + " was not declared in the keyspace.");
-      }
-
-      this.familyMap.put(fieldName, familyName);
-      this.columnMap.put(fieldName, columnName);
-      // TODO we should find a way of storing more values into this map
-      this.columnAttrMap.put(columnName, ttlValue!=null?ttlValue:DEFAULT_COLUMNS_TTL);
-    }
-  }
-
-  /**
-   * Add new column to the CassandraMapping using the the below parameters
-   * @param pFamilyName the column family name
-   * @param pFieldName the Avro field from the Schema
-   * @param pColumnName the column name within the column family.
-   */
-  public void addColumn(String pFamilyName, String pFieldName, String pColumnName){
-    this.familyMap.put(pFieldName, pFamilyName);
-    this.columnMap.put(pFieldName, pColumnName);
-  }
-
-  public String getFamily(String name) {
-    return this.familyMap.get(name);
-  }
-
-  public String getColumn(String name) {
-    return this.columnMap.get(name);
-  }
-
-  public Map<String,String> getFamilyMap(){
-    return this.familyMap;
-  }
-
-  public Map<String, String> getColumnsAttribs(){
-    return this.columnAttrMap;
-  }
-
-  /**
-   * Read family super attribute.
-   * @param family the family name
-   * @return true is the family is a super column family
-   */
-  public boolean isSuper(String family) {
-    return this.superFamilies.indexOf(family) != -1;
-  }
-
-  public List<ColumnFamilyDefinition> getColumnFamilyDefinitions() {
-    List<ColumnFamilyDefinition> list = new ArrayList<>();
-    for (String key: this.columnFamilyDefinitions.keySet()) {
-      ColumnFamilyDefinition columnFamilyDefinition = this.columnFamilyDefinitions.get(key);
-      ThriftCfDef thriftCfDef = new ThriftCfDef(columnFamilyDefinition);
-      list.add(thriftCfDef);
-    }
-    
-    return list;
-  }
-
-}

http://git-wip-us.apache.org/repos/asf/gora/blob/cbed41d0/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/CassandraMappingManager.java
----------------------------------------------------------------------
diff --git a/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/CassandraMappingManager.java b/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/CassandraMappingManager.java
deleted file mode 100644
index 6b46eec..0000000
--- a/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/CassandraMappingManager.java
+++ /dev/null
@@ -1,158 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.gora.cassandra.store;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-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;
-
-/**
- * A builder for creating the mapper.
- */
-public class CassandraMappingManager {
-  
-  public static final Logger LOG = LoggerFactory.getLogger(CassandraMappingManager.class);
-  
-  private static final String MAPPING_FILE = "gora-cassandra-mapping.xml";
-  private static final String KEYSPACE_ELEMENT = "keyspace";
-  private static final String NAME_ATTRIBUTE = "name";
-  private static final String MAPPING_ELEMENT = "class";
-  private static final String KEYCLASS_ATTRIBUTE = "keyClass";
-  private static final String HOST_ATTRIBUTE = "host";
-  private static final String CLUSTER_ATTRIBUTE = "cluster";
-  // singleton
-  private static CassandraMappingManager manager = new CassandraMappingManager();
-
-  public static CassandraMappingManager getManager() {
-    return manager;
-  }
-
- /**
-  * Objects to maintain mapped keyspaces
-  */
-  private Map<String, Element> keyspaceMap = null;
-  private Map<String, Element>  mappingMap = null;
-
-  private CassandraMappingManager() {
-    keyspaceMap = new HashMap<>();
-    mappingMap  = new HashMap<>();
-    try {
-      loadConfiguration();
-    }
-    catch (JDOMException | IOException e) {
-      LOG.error(e.toString());
-    }
-  }
-
-  public CassandraMapping get(Class<?> persistentClass) {
-    String className = persistentClass.getName();
-    Element mappingElement = mappingMap.get(className);
-    if (mappingElement == null) {
-      LOG.error("Mapping element does not exist for className=" + className);
-      return null;
-    }
-    String keyspaceName = mappingElement.getAttributeValue(KEYSPACE_ELEMENT);
-    if (LOG.isDebugEnabled()) {
-      LOG.debug("persistentClassName=" + className + " -> keyspaceName=" + keyspaceName);
-    }
-    Element keyspaceElement = keyspaceMap.get(keyspaceName);
-    if (keyspaceElement == null) {
-      LOG.error("Keyspace element does not exist for keyspaceName=" + keyspaceName);
-      return null;
-    }
-    return new CassandraMapping(keyspaceElement, mappingElement);
-  }
-
-  /**
-   * Primary class for loading Cassandra configuration from the 'MAPPING_FILE'.
-   * 
-   * @throws JDOMException
-   * @throws IOException
-   */
-  @SuppressWarnings("unchecked")
-  public void loadConfiguration() throws JDOMException, IOException {
-    SAXBuilder saxBuilder = new SAXBuilder();
-    // get mapping file
-    InputStream inputStream = getClass().getClassLoader().getResourceAsStream(MAPPING_FILE);
-    if (inputStream == null){
-      LOG.warn("Mapping file '" + MAPPING_FILE + "' could not be found!");
-      throw new IOException("Mapping file '" + MAPPING_FILE + "' could not be found!");
-    }
-    Document document = saxBuilder.build(inputStream);
-    if (document == null) {
-      LOG.warn("Mapping file '" + MAPPING_FILE + "' could not be found!");
-      throw new IOException("Mapping file '" + MAPPING_FILE + "' could not be found!");
-    }
-    Element root = document.getRootElement();
-    // find cassandra keyspace element
-    List<Element> keyspaces = root.getChildren(KEYSPACE_ELEMENT);
-    if (keyspaces == null || keyspaces.size() == 0) {
-      LOG.error("Error locating Cassandra Keyspace element!");
-    }
-    else {
-      for (Element keyspace : keyspaces) {
-        // log name, cluster and host for given keyspace(s)
-        String keyspaceName = keyspace.getAttributeValue(NAME_ATTRIBUTE);
-        String clusterName = keyspace.getAttributeValue(CLUSTER_ATTRIBUTE);
-        String hostName = keyspace.getAttributeValue(HOST_ATTRIBUTE);
-        if (LOG.isDebugEnabled()) {
-          LOG.debug("Located Cassandra Keyspace: '" + keyspaceName + "' in cluster '" + clusterName + 
-          "' on host '" + hostName + "'.");
-        }
-        if (keyspaceName == null) {
-          LOG.error("Error locating Cassandra Keyspace name attribute!");
-          continue;
-        }
-        keyspaceMap.put(keyspaceName, keyspace);
-      }
-    }
-      
-    // load column definitions    
-    List<Element> mappings = root.getChildren(MAPPING_ELEMENT);
-    if (mappings == null || mappings.size() == 0) {
-      LOG.error("Error locating Cassandra Mapping class element!");
-    }
-    else {
-      for (Element mapping : mappings) {
-        // associate persistent and class names for keyspace(s)
-        String className = mapping.getAttributeValue(NAME_ATTRIBUTE);
-        String keyClassName = mapping.getAttributeValue(KEYCLASS_ATTRIBUTE);
-        String keyspaceName = mapping.getAttributeValue(KEYSPACE_ELEMENT);
-        if (LOG.isDebugEnabled()) {
-        LOG.debug("Located Cassandra Mapping: keyClass: '" + keyClassName + "' in storage class '" 
-          + className + "' for Keyspace '" + keyspaceName + "'.");
-        }
-        if (className == null) {
-          LOG.error("Error locating Cassandra Mapping class name attribute!");
-          continue;
-        }
-        mappingMap.put(className, mapping);
-      }
-    }
-  }
-}

http://git-wip-us.apache.org/repos/asf/gora/blob/cbed41d0/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/CassandraStore.java
----------------------------------------------------------------------
diff --git a/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/CassandraStore.java b/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/CassandraStore.java
deleted file mode 100644
index 5f21aca..0000000
--- a/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/CassandraStore.java
+++ /dev/null
@@ -1,668 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.gora.cassandra.store;
-
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.Set;
-import java.util.Collections;
-import java.util.concurrent.ConcurrentHashMap;
-
-import me.prettyprint.hector.api.beans.ColumnSlice;
-import me.prettyprint.hector.api.beans.HColumn;
-import me.prettyprint.hector.api.beans.HSuperColumn;
-import me.prettyprint.hector.api.beans.Row;
-import me.prettyprint.hector.api.beans.SuperRow;
-import me.prettyprint.hector.api.beans.SuperSlice;
-
-import org.apache.avro.Schema;
-import org.apache.avro.Schema.Field;
-import org.apache.avro.Schema.Type;
-import org.apache.avro.generic.GenericArray;
-import org.apache.avro.generic.GenericData.Array;
-import org.apache.avro.io.BinaryEncoder;
-import org.apache.avro.specific.SpecificData;
-import org.apache.avro.specific.SpecificDatumWriter;
-import org.apache.gora.cassandra.query.CassandraQuery;
-import org.apache.gora.cassandra.query.CassandraResult;
-import org.apache.gora.cassandra.query.CassandraResultSet;
-import org.apache.gora.cassandra.query.CassandraRow;
-import org.apache.gora.cassandra.query.CassandraSubColumn;
-import org.apache.gora.cassandra.query.CassandraSuperColumn;
-import org.apache.gora.persistency.Persistent;
-import org.apache.gora.persistency.impl.DirtyListWrapper;
-import org.apache.gora.persistency.impl.PersistentBase;
-import org.apache.gora.query.PartitionQuery;
-import org.apache.gora.query.Query;
-import org.apache.gora.query.Result;
-import org.apache.gora.query.impl.PartitionQueryImpl;
-import org.apache.gora.store.DataStoreFactory;
-import org.apache.gora.store.impl.DataStoreBase;
-import org.apache.gora.cassandra.serializers.AvroSerializerUtil;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * {@link org.apache.gora.cassandra.store.CassandraStore} is the primary class
- * responsible for directing Gora CRUD operations into Cassandra. We (delegate) rely
- * heavily on {@link org.apache.gora.cassandra.store.CassandraClient} for many operations
- * such as initialization, creating and deleting schemas (Cassandra Keyspaces), etc.
- */
-public class CassandraStore<K, T extends PersistentBase> extends DataStoreBase<K, T> {
-
-  /** Logging implementation */
-  public static final Logger LOG = LoggerFactory.getLogger(CassandraStore.class);
-
-  /** Consistency property level for Cassandra column families */
-  private static final String COL_FAM_CL = "cf.consistency.level";
-   
-  /** Consistency property level for Cassandra read operations. */
-  private static final String READ_OP_CL = "read.consistency.level";
-  
-  /** Consistency property level for Cassandra write operations. */
-  private static final String WRITE_OP_CL = "write.consistency.level";
-  
-  /** Variables to hold different consistency levels defined by the properties. */
-  public static String colFamConsLvl;
-  public static String readOpConsLvl;
-  public static String writeOpConsLvl;
-  
-  private CassandraClient<K, T> cassandraClient = new CassandraClient<>();
-
-  /**
-   * Fixed string with value "UnionIndex" used to generate an extra column based on 
-   * the original field's name
-   */
-  public static final String UNION_COL_SUFIX = "_UnionIndex";
-
-  /**
-   * Default schema index with value "0" used when AVRO Union data types are stored
-   */
-  public static final int DEFAULT_UNION_SCHEMA = 0;
-
-  /**
-   * The values are Avro fields pending to be stored.
-   *
-   * We want to iterate over the keys in insertion order.
-   * We don't want to lock the entire collection before iterating over the keys, 
-   * since in the meantime other threads are adding entries to the map.
-   */
-  private Map<K, T> buffer = Collections.synchronizedMap(new LinkedHashMap<K, T>());
-
-  public static final ThreadLocal<BinaryEncoder> encoders =
-      new ThreadLocal<>();
-  
-  /**
-   * Create a {@link java.util.concurrent.ConcurrentHashMap} for the 
-   * datum readers and writers. 
-   * This is necessary because they are not thread safe, at least not before 
-   * Avro 1.4.0 (See AVRO-650).
-   * When they are thread safe, it is possible to maintain a single reader and
-   * writer pair for every schema, instead of one for every thread.
-   * @see <a href="https://issues.apache.org/jira/browse/AVRO-650">AVRO-650</a>
-   */
-  public static final ConcurrentHashMap<String, SpecificDatumWriter<?>> writerMap = 
-      new ConcurrentHashMap<>();
-  
-  /** The default constructor for CassandraStore */
-  public CassandraStore() throws Exception {
-  }
-
-  /** 
-   * Initialize is called when then the call to 
-   * {@link org.apache.gora.store.DataStoreFactory#createDataStore}
-   * is made. In this case, we merely delegate the store initialization to the 
-   * {@link org.apache.gora.cassandra.store.CassandraClient#initialize}.
-   *
-   * @param keyClass
-   * @param persistent
-   * @param properties
-   */
-  public void initialize(Class<K> keyClass, Class<T> persistent, Properties properties) {
-    try {
-      super.initialize(keyClass, persistent, properties);
-      if (autoCreateSchema) {
-        // If this is not set, then each Cassandra client should set its default
-        // column family
-        colFamConsLvl = DataStoreFactory.findProperty(properties, this, COL_FAM_CL, null);
-        // operations
-        readOpConsLvl = DataStoreFactory.findProperty(properties, this, READ_OP_CL, null);
-        writeOpConsLvl = DataStoreFactory.findProperty(properties, this, WRITE_OP_CL, null);
-      }
-      this.cassandraClient.initialize(keyClass, persistent, properties);
-    } catch (Exception e) {
-      LOG.error(e.getMessage(), e);
-    }
-  }
-
-  @Override
-  public void close() {
-    LOG.debug("close");
-    flush();
-  }
-
-  @Override
-  public void createSchema() {
-    LOG.debug("creating Cassandra keyspace");
-    this.cassandraClient.checkKeyspace();
-  }
-
-  @Override
-  public boolean delete(K key) {
-    this.cassandraClient.deleteByKey(key);
-    return true;
-  }
-
-  @Override
-  public long deleteByQuery(Query<K, T> query) {
-    LOG.debug("delete by query " + query);
-    return 0;
-  }
-
-  @Override
-  public void deleteSchema() {
-    LOG.debug("delete schema");
-    this.cassandraClient.dropKeyspace();
-  }
-
-  /**
-   * When executing Gora Queries in Cassandra we query the Cassandra keyspace by families.
-   * When we add sub/supercolumns, Gora keys are mapped to Cassandra partition keys only. 
-   * This is because we follow the Cassandra logic where column family data is 
-   * partitioned across nodes based on row Key.
-   */
-  @Override
-  public Result<K, T> execute(Query<K, T> query) {
-
-    Map<String, List<String>> familyMap = this.cassandraClient.getFamilyMap(query);
-    Map<String, String> reverseMap = this.cassandraClient.getReverseMap(query);
-
-    CassandraQuery<K, T> cassandraQuery = new CassandraQuery<>();
-    cassandraQuery.setQuery(query);
-    cassandraQuery.setFamilyMap(familyMap);
-
-    CassandraResult<K, T> cassandraResult = new CassandraResult<>(this, query);
-    cassandraResult.setReverseMap(reverseMap);
-
-    CassandraResultSet<K> cassandraResultSet = new CassandraResultSet<>();
-
-    // We query Cassandra keyspace by families.
-    for (String family : familyMap.keySet()) {
-      if (family == null) {
-        continue;
-      }
-      if (this.cassandraClient.isSuper(family)) {
-        addSuperColumns(family, cassandraQuery, cassandraResultSet);
-
-      } else {
-        addSubColumns(family, cassandraQuery, cassandraResultSet);
-      }
-    }
-
-    cassandraResult.setResultSet(cassandraResultSet);
-
-    return cassandraResult;
-  }
-
-  /**
-   * When we add subcolumns, Gora keys are mapped to Cassandra partition keys only. 
-   * This is because we follow the Cassandra logic where column family data is 
-   * partitioned across nodes based on row Key.
-   */
-  private void addSubColumns(String family, CassandraQuery<K, T> cassandraQuery,
-      CassandraResultSet<K> cassandraResultSet) {
-    // select family columns that are included in the query
-    List<Row<K, ByteBuffer, ByteBuffer>> rows = this.cassandraClient.execute(cassandraQuery, family);
-
-    for (Row<K, ByteBuffer, ByteBuffer> row : rows) {
-      K key = row.getKey();
-
-      // find associated row in the resultset
-      CassandraRow<K> cassandraRow = cassandraResultSet.getRow(key);
-      if (cassandraRow == null) {
-        cassandraRow = new CassandraRow<>();
-        cassandraResultSet.putRow(key, cassandraRow);
-        cassandraRow.setKey(key);
-      }
-
-      ColumnSlice<ByteBuffer, ByteBuffer> columnSlice = row.getColumnSlice();
-
-      for (HColumn<ByteBuffer, ByteBuffer> hColumn : columnSlice.getColumns()) {
-        CassandraSubColumn cassandraSubColumn = new CassandraSubColumn();
-        cassandraSubColumn.setValue(hColumn);
-        cassandraSubColumn.setFamily(family);
-        cassandraRow.add(cassandraSubColumn);
-      }
-
-    }
-  }
-
-  /**
-   * When we add supercolumns, Gora keys are mapped to Cassandra partition keys only. 
-   * This is because we follow the Cassandra logic where column family data is 
-   * partitioned across nodes based on row Key.
-   */
-  private void addSuperColumns(String family, CassandraQuery<K, T> cassandraQuery, 
-      CassandraResultSet<K> cassandraResultSet) {
-
-    List<SuperRow<K, String, ByteBuffer, ByteBuffer>> superRows = this.cassandraClient.executeSuper(cassandraQuery, family);
-    for (SuperRow<K, String, ByteBuffer, ByteBuffer> superRow: superRows) {
-      K key = superRow.getKey();
-      CassandraRow<K> cassandraRow = cassandraResultSet.getRow(key);
-      if (cassandraRow == null) {
-        cassandraRow = new CassandraRow<>();
-        cassandraResultSet.putRow(key, cassandraRow);
-        cassandraRow.setKey(key);
-      }
-
-      SuperSlice<String, ByteBuffer, ByteBuffer> superSlice = superRow.getSuperSlice();
-      for (HSuperColumn<String, ByteBuffer, ByteBuffer> hSuperColumn: superSlice.getSuperColumns()) {
-        CassandraSuperColumn cassandraSuperColumn = new CassandraSuperColumn();
-        cassandraSuperColumn.setValue(hSuperColumn);
-        cassandraSuperColumn.setFamily(family);
-        cassandraRow.add(cassandraSuperColumn);
-      }
-    }
-  }
-
-  /**
-   * Flush the buffer which is a synchronized {@link java.util.LinkedHashMap}
-   * storing fields pending to be stored by 
-   * {@link org.apache.gora.cassandra.store.CassandraStore#put(Object, PersistentBase)}
-   * operations. Invoking this method therefore writes the buffered rows
-   * into Cassandra.
-   * @see org.apache.gora.store.DataStore#flush()
-   */
-  @Override
-  public void flush() {
-
-    Set<K> keys = this.buffer.keySet();
-
-    // this duplicates memory footprint
-    @SuppressWarnings("unchecked")
-    K[] keyArray = (K[]) keys.toArray();
-
-    // iterating over the key set directly would throw 
-    //ConcurrentModificationException with java.util.HashMap and subclasses
-    for (K key: keyArray) {
-      T value = this.buffer.get(key);
-      if (value == null) {
-        LOG.info("Value to update is null for key: " + key);
-        continue;
-      }
-      Schema schema = value.getSchema();
-
-      for (Field field: schema.getFields()) {
-        if (value.isDirty(field.pos())) {
-          addOrUpdateField(key, field, field.schema(), value.get(field.pos()));
-        }
-      }
-    }
-
-    // remove flushed rows from the buffer as all 
-    // added or updated fields should now have been written.
-    for (K key: keyArray) {
-      this.buffer.remove(key);
-    }
-  }
-
-  @Override
-  public T get(K key, String[] fields) {
-    CassandraQuery<K,T> query = new CassandraQuery<>();
-    query.setDataStore(this);
-    query.setKeyRange(key, key);
-    
-    if (fields == null){
-      fields = this.getFields();
-    }
-    query.setFields(fields);
-
-    query.setLimit(1);
-    Result<K,T> result = execute(query);
-    boolean hasResult = false;
-    try {
-      hasResult = result.next();
-    } catch (Exception e) {
-      LOG.error(e.getMessage());
-      throw new RuntimeException(e);
-    }
-    return hasResult ? result.get() : null;
-  }
-
-  @Override
-  public List<PartitionQuery<K, T>> getPartitions(Query<K, T> query)
-      throws IOException {
-    // TODO GORA-298 Implement CassandraStore#getPartitions
-    List<PartitionQuery<K,T>> partitions = new ArrayList<>();
-    PartitionQueryImpl<K, T> pqi = new PartitionQueryImpl<>(query);
-    pqi.setConf(getConf());
-    partitions.add(pqi);
-    return partitions;
-  }
-
-  /**
-   * In Cassandra Schemas are referred to as Keyspaces
-   * @return Keyspace
-   */
-  @Override
-  public String getSchemaName() {
-    return this.cassandraClient.getKeyspaceName();
-  }
-
-  @Override
-  public Query<K, T> newQuery() {
-    Query<K,T> query = new CassandraQuery<>(this);
-    query.setFields(getFieldsToQuery(null));
-    return query;
-  }
-
-  /**
-   * When doing the
-   * {@link org.apache.gora.cassandra.store.CassandraStore#put(Object, PersistentBase)}
-   * operation, the logic is as follows:
-   * <ol>
-   * <li>Obtain the Avro {@link org.apache.avro.Schema} for the object.</li>
-   * <li>Create a new duplicate instance of the object (explained in more detail below) **.</li>
-   * <li>Obtain a {@link java.util.List} of the {@link org.apache.avro.Schema}
-   * {@link org.apache.avro.Schema.Field}'s.</li>
-   * <li>Iterate through the field {@link java.util.List}. This allows us to
-   * consequently process each item.</li>
-   * <li>Check to see if the {@link org.apache.avro.Schema.Field} is NOT dirty.
-   * If this condition is true then we DO NOT process this field.</li>
-   * <li>Obtain the element at the specified position in this list so we can
-   * directly operate on it.</li>
-   * <li>Obtain the {@link org.apache.avro.Schema.Type} of the element obtained
-   * above and process it accordingly. N.B. For nested type ARRAY, MAP
-   * RECORD or UNION, we shadow the checks in bullet point 5 above to infer that the
-   * {@link org.apache.avro.Schema.Field} is either at
-   * position 0 OR it is NOT dirty. If one of these conditions is true then we DO NOT
-   * process this field. This is carried out in
-   * {@link org.apache.gora.cassandra.store.CassandraStore#getFieldValue(Schema, Type, Object)}</li>
-   * <li>We then insert the Key and Object into the {@link java.util.LinkedHashMap} buffer
-   * before being flushed. This performs a structural modification of the map.</li>
-   * </ol>
-   * ** We create a duplicate instance of the object to be persisted and insert processed
-   * objects into a synchronized {@link java.util.LinkedHashMap}. This allows
-   * us to keep all the objects in memory till flushing.
-   *
-   * @param key   for the Avro Record (object).
-   * @param value Record object to be persisted in Cassandra
-   * @see org.apache.gora.store.DataStore#put(java.lang.Object,org.apache.gora.persistency.Persistent)
-   */
-  @Override
-  public void put(K key, T value) {
-    Schema schema = value.getSchema();
-    @SuppressWarnings("unchecked")
-    T p = (T) SpecificData.get().newRecord(value, schema);
-    List<Field> fields = schema.getFields();
-    for (int i = 1; i < fields.size(); i++) {
-      if (!value.isDirty(i)) {
-        continue;
-      }
-      Field field = fields.get(i);
-      Type type = field.schema().getType();
-      Object fieldValue = value.get(field.pos());
-      Schema fieldSchema = field.schema();
-      // check if field has a nested structure (array, map, record or union)
-      fieldValue = getFieldValue(fieldSchema, type, fieldValue);
-      p.put(field.pos(), fieldValue);
-    }
-    // this performs a structural modification of the map
-    this.buffer.put(key, p);
-  }
-
-  /**
-   * For every field within an object, we pass in a field schema, Type and value.
-   * This enables us to process fields (based on their characteristics) 
-   * preparing them for persistence.
-   * @param fieldSchema the associated field schema
-   * @param type the field type
-   * @param fieldValue the field value.
-   * @return
-   */
-  private Object getFieldValue(Schema fieldSchema, Type type, Object fieldValue ){
-    switch(type) {
-    case RECORD:
-      PersistentBase persistent = (PersistentBase) fieldValue;
-      PersistentBase newRecord = (PersistentBase) SpecificData.get().newRecord(persistent, persistent.getSchema());
-      for (Field member: fieldSchema.getFields()) {
-        if (member.pos() == 0 || !persistent.isDirty()) {
-          continue;
-        }
-        Schema memberSchema = member.schema();
-        Type memberType = memberSchema.getType();
-        Object memberValue = persistent.get(member.pos());
-        newRecord.put(member.pos(), getFieldValue(memberSchema, memberType, memberValue));
-      }
-      fieldValue = newRecord;
-      break;
-    case MAP:
-      Map<?, ?> map = (Map<?, ?>) fieldValue;
-      fieldValue = map;
-      break;
-    case ARRAY:
-      fieldValue = (List<?>) fieldValue;
-      break;
-    case UNION:
-      // storing the union selected schema, the actual value will 
-      // be stored as soon as we get break out.
-      if (fieldValue != null){
-        int schemaPos = getUnionSchema(fieldValue,fieldSchema);
-        Schema unionSchema = fieldSchema.getTypes().get(schemaPos);
-        Type unionType = unionSchema.getType();
-        fieldValue = getFieldValue(unionSchema, unionType, fieldValue);
-      }
-      //p.put( schemaPos, p.getSchema().getField(field.name() + CassandraStore.UNION_COL_SUFIX));
-      //p.put(fieldPos, fieldValue);
-      break;
-    default:
-      break;
-    }    
-    return fieldValue;
-  }
-  
-  /**
-   * Add a field to Cassandra according to its type.
-   * @param key     the key of the row where the field should be added
-   * @param field   the Avro field representing a datum
-   * @param schema  the schema belonging to the particular Avro field
-   * @param value   the field value
-   */
-  @SuppressWarnings({ "unchecked", "rawtypes" })
-  private void addOrUpdateField(K key, Field field, Schema schema, Object value) {
-    Type type = schema.getType();
-    // checking if the value to be updated is used for saving union schema
-    if (!field.name().contains(CassandraStore.UNION_COL_SUFIX)){
-      switch (type) {
-      case STRING:
-      case BOOLEAN:
-      case INT:
-      case LONG:
-      case BYTES:
-      case FLOAT:
-      case DOUBLE:
-      case FIXED:
-        this.cassandraClient.addColumn(key, field.name(), value);
-        break;
-      case RECORD:
-        if (value != null) {
-          if (value instanceof PersistentBase) {
-            PersistentBase persistentBase = (PersistentBase) value;            
-            try {
-              byte[] byteValue = AvroSerializerUtil.serializer(persistentBase, schema);
-              this.cassandraClient.addColumn(key, field.name(), byteValue);
-            } catch (IOException e) {
-              LOG.warn(field.name() + " named record could not be serialized.");
-            }
-          } else {
-            LOG.warn("Record with value: " + value.toString() + " not supported for field: " + field.name());
-          }
-        } else {
-          LOG.warn("Setting content of: " + field.name() + " to null.");
-          String familyName =  this.cassandraClient.getCassandraMapping().getFamily(field.name());
-          this.cassandraClient.deleteColumn(key, familyName, this.cassandraClient.toByteBuffer(field.name()));
-        }
-        break;
-      case MAP:
-        if (value != null) {
-          if (value instanceof Map<?, ?>) {            
-            Map<CharSequence,Object> map = (Map<CharSequence,Object>)value;
-            Schema valueSchema = schema.getValueType();
-            Type valueType = valueSchema.getType();
-            if (Type.UNION.equals(valueType)){
-              Map<CharSequence,Object> valueMap = new HashMap<>();
-              for (CharSequence mapKey: map.keySet()) {
-                Object mapValue = map.get(mapKey);
-                int valueUnionIndex = getUnionSchema(mapValue, valueSchema);
-                valueMap.put((mapKey+UNION_COL_SUFIX), valueUnionIndex);
-                valueMap.put(mapKey, mapValue);
-              }
-              map = valueMap;
-            }
-            
-            String familyName = this.cassandraClient.getCassandraMapping().getFamily(field.name());
-            
-            // If map is not super column. We using Avro serializer. 
-            if (!this.cassandraClient.isSuper( familyName )){
-              try {
-                byte[] byteValue = AvroSerializerUtil.serializer(map, schema);
-                this.cassandraClient.addColumn(key, field.name(), byteValue);
-              } catch (IOException e) {
-                LOG.warn(field.name() + " named map could not be serialized.");
-              }
-            }else{
-              this.cassandraClient.addStatefulHashMap(key, field.name(), map);              
-            }
-          } else {
-            LOG.warn("Map with value: " + value.toString() + " not supported for field: " + field.name());
-          }
-        } else {
-          // delete map
-          LOG.warn("Setting content of: " + field.name() + " to null.");
-          this.cassandraClient.deleteStatefulHashMap(key, field.name());
-        }
-        break;
-      case ARRAY:
-        if (value != null) {
-          if (value instanceof DirtyListWrapper<?>) {
-            DirtyListWrapper fieldValue = (DirtyListWrapper<?>)value;
-            GenericArray valueArray = new Array(fieldValue.size(), schema);
-            for (int i = 0; i < fieldValue.size(); i++) {
-              valueArray.add(i, fieldValue.get(i));
-            }
-            this.cassandraClient.addGenericArray(key, field.name(), (GenericArray<?>)valueArray);
-          } else {
-            LOG.warn("Array with value: " + value.toString() + " not supported for field: " + field.name());
-          }
-        } else {
-          LOG.warn("Setting content of: " + field.name() + " to null.");
-          this.cassandraClient.deleteGenericArray(key, field.name());
-        }
-        break;
-      case UNION:
-     // adding union schema index
-        String columnName = field.name() + UNION_COL_SUFIX;
-        String familyName = this.cassandraClient.getCassandraMapping().getFamily(field.name());
-        if(value != null) {
-          int schemaPos = getUnionSchema(value, schema);
-          LOG.debug("Union with value: " + value.toString() + " at index: " + schemaPos + " supported for field: " + field.name());
-          this.cassandraClient.getCassandraMapping().addColumn(familyName, columnName, columnName);
-          if (this.cassandraClient.isSuper( familyName )){
-            this.cassandraClient.addSubColumn(key, columnName, columnName, schemaPos);
-          }else{
-            this.cassandraClient.addColumn(key, columnName, schemaPos);
-            
-          }
-          //this.cassandraClient.getCassandraMapping().addColumn(familyName, columnName, columnName);
-          // adding union value
-          Schema unionSchema = schema.getTypes().get(schemaPos);
-          addOrUpdateField(key, field, unionSchema, value);
-          //this.cassandraClient.addColumn(key, field.name(), value);
-        } else {
-          LOG.warn("Setting content of: " + field.name() + " to null.");
-          if (this.cassandraClient.isSuper( familyName )){
-            this.cassandraClient.deleteSubColumn(key, field.name());
-          } else {
-            this.cassandraClient.deleteColumn(key, familyName, this.cassandraClient.toByteBuffer(field.name()));
-          }
-        }
-        break;
-      default:
-        LOG.warn("Type: " + type.name() + " not considered for field: " + field.name() + ". Please report this to dev@gora.apache.org");
-      }
-    }
-  }
-
-  /**
-   * Given an object and the object schema this function obtains,
-   * from within the UNION schema, the position of the type used.
-   * If no data type can be inferred then we return a default value
-   * of position 0.
-   * @param pValue
-   * @param pUnionSchema
-   * @return the unionSchemaPosition.
-   */
-  private int getUnionSchema(Object pValue, Schema pUnionSchema){
-    int unionSchemaPos = 0;
-//    String valueType = pValue.getClass().getSimpleName();
-    for (Schema currentSchema : pUnionSchema.getTypes()) {
-      Type schemaType = currentSchema.getType();
-      if (pValue instanceof CharSequence && schemaType.equals(Type.STRING))
-        return unionSchemaPos;
-      else if (pValue instanceof ByteBuffer && schemaType.equals(Type.BYTES))
-        return unionSchemaPos;
-      else if (pValue instanceof Integer && schemaType.equals(Type.INT))
-        return unionSchemaPos;
-      else if (pValue instanceof Long && schemaType.equals(Type.LONG))
-        return unionSchemaPos;
-      else if (pValue instanceof Double && schemaType.equals(Type.DOUBLE))
-        return unionSchemaPos;
-      else if (pValue instanceof Float && schemaType.equals(Type.FLOAT))
-        return unionSchemaPos;
-      else if (pValue instanceof Boolean && schemaType.equals(Type.BOOLEAN))
-        return unionSchemaPos;
-      else if (pValue instanceof Map && schemaType.equals(Type.MAP))
-        return unionSchemaPos;
-      else if (pValue instanceof List && schemaType.equals(Type.ARRAY))
-        return unionSchemaPos;
-      else if (pValue instanceof Persistent && schemaType.equals(Type.RECORD))
-        return unionSchemaPos;
-      unionSchemaPos++;
-    }
-    // if we weren't able to determine which data type it is, then we return the default
-    return DEFAULT_UNION_SCHEMA;
-  }
-
-  /**
-   * Simple method to check if a Cassandra Keyspace exists.
-   * @return true if a Keyspace exists.
-   */
-  @Override
-  public boolean schemaExists() {
-    LOG.info("schema exists");
-    return cassandraClient.keyspaceExists();
-  }
-
-}

http://git-wip-us.apache.org/repos/asf/gora/blob/cbed41d0/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/HectorUtils.java
----------------------------------------------------------------------
diff --git a/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/HectorUtils.java b/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/HectorUtils.java
deleted file mode 100644
index 3f33202..0000000
--- a/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/HectorUtils.java
+++ /dev/null
@@ -1,117 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.gora.cassandra.store;
-
-import java.nio.ByteBuffer;
-import java.util.Arrays;
-
-import me.prettyprint.cassandra.serializers.ByteBufferSerializer;
-import me.prettyprint.cassandra.serializers.IntegerSerializer;
-import me.prettyprint.cassandra.serializers.StringSerializer;
-import me.prettyprint.hector.api.beans.HColumn;
-import me.prettyprint.hector.api.beans.HSuperColumn;
-import me.prettyprint.hector.api.factory.HFactory;
-import me.prettyprint.hector.api.mutation.Mutator;
-
-import org.apache.gora.persistency.Persistent;
-
-/**
- * This class it not thread safe.
- * According to Hector's JavaDoc a Mutator isn't thread safe, too.
- * @see CassandraClient for safe usage.
- */
-public class HectorUtils<K,T extends Persistent> {
-
-  public static<K> void insertColumn(Mutator<K> mutator, K key, String columnFamily, ByteBuffer columnName, ByteBuffer columnValue, String ttlAttr) {
-    mutator.insert(key, columnFamily, createColumn(columnName, columnValue, ttlAttr));
-  }
-
-  public static<K> void insertColumn(Mutator<K> mutator, K key, String columnFamily, String columnName, ByteBuffer columnValue, String ttlAttr) {
-    mutator.insert(key, columnFamily, createColumn(columnName, columnValue, ttlAttr));
-  }
-
-
-  public static<K> HColumn<ByteBuffer,ByteBuffer> createColumn(ByteBuffer name, ByteBuffer value, String ttlAttr) {
-	int ttl = Integer.parseInt(ttlAttr);
-	HColumn<ByteBuffer,ByteBuffer> col = HFactory.createColumn(name, value, ByteBufferSerializer.get(), ByteBufferSerializer.get());
-	
-	if( 0 < ttl ) {
-		col.setTtl( ttl ); 
-	}
-	
-	return col;
-  }
-
-  public static<K> HColumn<String,ByteBuffer> createColumn(String name, ByteBuffer value, String ttlAttr) {
-	int ttl = Integer.parseInt(ttlAttr);
-	HColumn<String,ByteBuffer> col = HFactory.createColumn(name, value, StringSerializer.get(), ByteBufferSerializer.get());
-	
-	if( 0 < ttl ) {
-		col.setTtl( ttl );
-	}
-	
-	return col;
-  }
-
-  public static<K> HColumn<Integer,ByteBuffer> createColumn(Integer name, ByteBuffer value, String ttlAttr) {
-    int ttl = Integer.parseInt(ttlAttr);
-    HColumn<Integer,ByteBuffer> col = HFactory.createColumn(name, value, IntegerSerializer.get(), ByteBufferSerializer.get());
-    
-    if( 0 < ttl ) {
-    	col.setTtl( ttl );
-    }
-    
-	return col;
-  }
-
-
-  public static<K> void insertSubColumn(Mutator<K> mutator, K key, String columnFamily, String superColumnName, ByteBuffer columnName, ByteBuffer columnValue, String ttlAttr) {
-    mutator.insert(key, columnFamily, createSuperColumn(superColumnName, columnName, columnValue, ttlAttr));
-  }
-
-  public static<K> void insertSubColumn(Mutator<K> mutator, K key, String columnFamily, String superColumnName, String columnName, ByteBuffer columnValue, String ttlAttr) {
-    mutator.insert(key, columnFamily, createSuperColumn(superColumnName, columnName, columnValue, ttlAttr));
-  }
-
-  public static<K> void insertSubColumn(Mutator<K> mutator, K key, String columnFamily, String superColumnName, Integer columnName, ByteBuffer columnValue, String ttlAttr) {
-    mutator.insert(key, columnFamily, createSuperColumn(superColumnName, columnName, columnValue, ttlAttr));
-  }
-
-
-  public static<K> void deleteSubColumn(Mutator<K> mutator, K key, String columnFamily, String superColumnName, ByteBuffer columnName) {
-    mutator.subDelete(key, columnFamily, superColumnName, columnName, StringSerializer.get(), ByteBufferSerializer.get());
-  }
-
-  public static<K> void deleteColumn(Mutator<K> mutator, K key, String columnFamily, ByteBuffer columnName){
-    mutator.delete(key, columnFamily, columnName, ByteBufferSerializer.get());
-  }
-
-  public static<K> HSuperColumn<String,ByteBuffer,ByteBuffer> createSuperColumn(String superColumnName, ByteBuffer columnName, ByteBuffer columnValue, String ttlAttr) {
-    return HFactory.createSuperColumn(superColumnName, Arrays.asList(createColumn(columnName, columnValue, ttlAttr)), StringSerializer.get(), ByteBufferSerializer.get(), ByteBufferSerializer.get());
-  }
-
-  public static<K> HSuperColumn<String,String,ByteBuffer> createSuperColumn(String superColumnName, String columnName, ByteBuffer columnValue, String ttlAttr) {
-    return HFactory.createSuperColumn(superColumnName, Arrays.asList(createColumn(columnName, columnValue, ttlAttr)), StringSerializer.get(), StringSerializer.get(), ByteBufferSerializer.get());
-  }
-
-  public static<K> HSuperColumn<String,Integer,ByteBuffer> createSuperColumn(String superColumnName, Integer columnName, ByteBuffer columnValue, String ttlAttr) {
-    return HFactory.createSuperColumn(superColumnName, Arrays.asList(createColumn(columnName, columnValue, ttlAttr)), StringSerializer.get(), IntegerSerializer.get(), ByteBufferSerializer.get());
-  }
-
-}


Mime
View raw message