gora-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rmarroq...@apache.org
Subject svn commit: r1586888 [1/10] - in /gora/trunk: ./ bin/ gora-accumulo/ gora-accumulo/src/main/java/org/apache/gora/accumulo/encoders/ gora-accumulo/src/main/java/org/apache/gora/accumulo/query/ gora-accumulo/src/main/java/org/apache/gora/accumulo/store/ ...
Date Sat, 12 Apr 2014 19:21:56 GMT
Author: rmarroquin
Date: Sat Apr 12 19:21:53 2014
New Revision: 1586888

URL: http://svn.apache.org/r1586888
Log:
GORA-321. Merge GORA_94 into Gora trunk

Added:
    gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/AvroSerializerUtil.java   (with props)
    gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/CharSequenceSerializer.java   (with props)
    gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/ListSerializer.java   (with props)
    gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/MapSerializer.java   (with props)
    gora/trunk/gora-compiler/
    gora/trunk/gora-compiler-cli/
    gora/trunk/gora-compiler-cli/pom.xml   (with props)
    gora/trunk/gora-compiler-cli/src/
    gora/trunk/gora-compiler-cli/src/main/
    gora/trunk/gora-compiler-cli/src/main/java/
    gora/trunk/gora-compiler-cli/src/main/java/org/
    gora/trunk/gora-compiler-cli/src/main/java/org/apache/
    gora/trunk/gora-compiler-cli/src/main/java/org/apache/gora/
    gora/trunk/gora-compiler-cli/src/main/java/org/apache/gora/compiler/
    gora/trunk/gora-compiler-cli/src/main/java/org/apache/gora/compiler/cli/
    gora/trunk/gora-compiler-cli/src/main/java/org/apache/gora/compiler/cli/GoraCompilerCLI.java   (with props)
    gora/trunk/gora-compiler/pom.xml   (with props)
    gora/trunk/gora-compiler/src/
    gora/trunk/gora-compiler/src/main/
    gora/trunk/gora-compiler/src/main/java/
    gora/trunk/gora-compiler/src/main/java/org/
    gora/trunk/gora-compiler/src/main/java/org/apache/
    gora/trunk/gora-compiler/src/main/java/org/apache/gora/
    gora/trunk/gora-compiler/src/main/java/org/apache/gora/compiler/
    gora/trunk/gora-compiler/src/main/java/org/apache/gora/compiler/GoraCompiler.java   (with props)
    gora/trunk/gora-compiler/src/main/velocity/
    gora/trunk/gora-compiler/src/main/velocity/org/
    gora/trunk/gora-compiler/src/main/velocity/org/apache/
    gora/trunk/gora-compiler/src/main/velocity/org/apache/gora/
    gora/trunk/gora-compiler/src/main/velocity/org/apache/gora/compiler/
    gora/trunk/gora-compiler/src/main/velocity/org/apache/gora/compiler/templates/
    gora/trunk/gora-compiler/src/main/velocity/org/apache/gora/compiler/templates/record.vm
    gora/trunk/gora-core/src/examples/avro/immutable_fields.json
    gora/trunk/gora-core/src/examples/java/org/apache/gora/examples/generated/ImmutableFields.java   (with props)
    gora/trunk/gora-core/src/examples/java/org/apache/gora/examples/generated/V2.java   (with props)
    gora/trunk/gora-core/src/main/java/org/apache/gora/persistency/Dirtyable.java   (with props)
    gora/trunk/gora-core/src/main/java/org/apache/gora/persistency/Tombstone.java   (with props)
    gora/trunk/gora-core/src/main/java/org/apache/gora/persistency/Tombstones.java   (with props)
    gora/trunk/gora-core/src/main/java/org/apache/gora/persistency/impl/DirtyCollectionWrapper.java   (with props)
    gora/trunk/gora-core/src/main/java/org/apache/gora/persistency/impl/DirtyFlag.java   (with props)
    gora/trunk/gora-core/src/main/java/org/apache/gora/persistency/impl/DirtyIteratorWrapper.java   (with props)
    gora/trunk/gora-core/src/main/java/org/apache/gora/persistency/impl/DirtyListIterator.java   (with props)
    gora/trunk/gora-core/src/main/java/org/apache/gora/persistency/impl/DirtyListWrapper.java   (with props)
    gora/trunk/gora-core/src/main/java/org/apache/gora/persistency/impl/DirtyMapWrapper.java   (with props)
    gora/trunk/gora-core/src/main/java/org/apache/gora/persistency/impl/DirtySetWrapper.java   (with props)
    gora/trunk/gora-core/src/test/conf/hadoop-metrics2.properties   (with props)
    gora/trunk/gora-core/src/test/java/org/apache/gora/examples/
    gora/trunk/gora-core/src/test/java/org/apache/gora/examples/TestWebPageDataCreator.java   (with props)
Modified:
    gora/trunk/CHANGES.txt
    gora/trunk/bin/gora
    gora/trunk/gora-accumulo/pom.xml
    gora/trunk/gora-accumulo/src/main/java/org/apache/gora/accumulo/encoders/BinaryEncoder.java
    gora/trunk/gora-accumulo/src/main/java/org/apache/gora/accumulo/query/AccumuloResult.java
    gora/trunk/gora-accumulo/src/main/java/org/apache/gora/accumulo/store/AccumuloStore.java
    gora/trunk/gora-accumulo/src/test/resources/gora-accumulo-mapping.xml
    gora/trunk/gora-accumulo/src/test/resources/gora.properties
    gora/trunk/gora-cassandra/pom.xml
    gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraColumn.java
    gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraResult.java
    gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraSubColumn.java
    gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraSuperColumn.java
    gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/GenericArraySerializer.java
    gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/GoraSerializerTypeInferer.java
    gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/SpecificFixedSerializer.java
    gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/StatefulHashMapSerializer.java
    gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/TypeUtils.java
    gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/Utf8Serializer.java
    gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/CassandraClient.java
    gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/CassandraMapping.java
    gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/CassandraMappingManager.java
    gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/CassandraStore.java
    gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/HectorUtils.java
    gora/trunk/gora-cassandra/src/test/conf/cassandra.yaml
    gora/trunk/gora-cassandra/src/test/conf/gora-cassandra-mapping.xml
    gora/trunk/gora-cassandra/src/test/conf/gora.properties
    gora/trunk/gora-cassandra/src/test/conf/log4j-server.properties
    gora/trunk/gora-cassandra/src/test/java/org/apache/gora/cassandra/GoraCassandraTestDriver.java
    gora/trunk/gora-cassandra/src/test/java/org/apache/gora/cassandra/store/TestCassandraStore.java
    gora/trunk/gora-core/pom.xml
    gora/trunk/gora-core/src/examples/avro/employee.json
    gora/trunk/gora-core/src/examples/avro/tokendatum.json
    gora/trunk/gora-core/src/examples/avro/webpage.json
    gora/trunk/gora-core/src/examples/java/org/apache/gora/examples/WebPageDataCreator.java
    gora/trunk/gora-core/src/examples/java/org/apache/gora/examples/generated/Employee.java
    gora/trunk/gora-core/src/examples/java/org/apache/gora/examples/generated/Metadata.java
    gora/trunk/gora-core/src/examples/java/org/apache/gora/examples/generated/TokenDatum.java
    gora/trunk/gora-core/src/examples/java/org/apache/gora/examples/generated/WebPage.java
    gora/trunk/gora-core/src/main/java/org/apache/gora/avro/PersistentDatumReader.java
    gora/trunk/gora-core/src/main/java/org/apache/gora/avro/PersistentDatumWriter.java
    gora/trunk/gora-core/src/main/java/org/apache/gora/avro/store/AvroStore.java
    gora/trunk/gora-core/src/main/java/org/apache/gora/avro/store/DataFileAvroStore.java
    gora/trunk/gora-core/src/main/java/org/apache/gora/filter/MapFieldValueFilter.java
    gora/trunk/gora-core/src/main/java/org/apache/gora/filter/SingleFieldValueFilter.java
    gora/trunk/gora-core/src/main/java/org/apache/gora/mapreduce/FakeResolvingDecoder.java
    gora/trunk/gora-core/src/main/java/org/apache/gora/mapreduce/GoraMapReduceUtils.java
    gora/trunk/gora-core/src/main/java/org/apache/gora/mapreduce/PersistentDeserializer.java
    gora/trunk/gora-core/src/main/java/org/apache/gora/mapreduce/PersistentNonReusingSerialization.java
    gora/trunk/gora-core/src/main/java/org/apache/gora/mapreduce/PersistentSerialization.java
    gora/trunk/gora-core/src/main/java/org/apache/gora/mapreduce/PersistentSerializer.java
    gora/trunk/gora-core/src/main/java/org/apache/gora/memory/store/MemStore.java
    gora/trunk/gora-core/src/main/java/org/apache/gora/persistency/ListGenericArray.java
    gora/trunk/gora-core/src/main/java/org/apache/gora/persistency/Persistent.java
    gora/trunk/gora-core/src/main/java/org/apache/gora/persistency/State.java
    gora/trunk/gora-core/src/main/java/org/apache/gora/persistency/StateManager.java
    gora/trunk/gora-core/src/main/java/org/apache/gora/persistency/StatefulHashMap.java
    gora/trunk/gora-core/src/main/java/org/apache/gora/persistency/StatefulMap.java
    gora/trunk/gora-core/src/main/java/org/apache/gora/persistency/impl/BeanFactoryImpl.java
    gora/trunk/gora-core/src/main/java/org/apache/gora/persistency/impl/PersistentBase.java
    gora/trunk/gora-core/src/main/java/org/apache/gora/persistency/impl/StateManagerImpl.java
    gora/trunk/gora-core/src/main/java/org/apache/gora/persistency/ws/impl/BeanFactoryWSImpl.java
    gora/trunk/gora-core/src/main/java/org/apache/gora/persistency/ws/impl/PersistentWSBase.java
    gora/trunk/gora-core/src/main/java/org/apache/gora/persistency/ws/impl/StateManagerWSImpl.java
    gora/trunk/gora-core/src/main/java/org/apache/gora/store/DataStoreFactory.java
    gora/trunk/gora-core/src/main/java/org/apache/gora/store/impl/DataStoreBase.java
    gora/trunk/gora-core/src/main/java/org/apache/gora/util/AvroUtils.java
    gora/trunk/gora-core/src/main/java/org/apache/gora/util/ByteUtils.java
    gora/trunk/gora-core/src/main/java/org/apache/gora/util/IOUtils.java
    gora/trunk/gora-core/src/main/java/org/apache/gora/util/ReflectionUtils.java
    gora/trunk/gora-core/src/test/java/org/apache/gora/GoraTestDriver.java
    gora/trunk/gora-core/src/test/java/org/apache/gora/avro/TestPersistentDatumReader.java
    gora/trunk/gora-core/src/test/java/org/apache/gora/filter/TestMapFieldValueFilter.java
    gora/trunk/gora-core/src/test/java/org/apache/gora/filter/TestSingleFieldValueFilter.java
    gora/trunk/gora-core/src/test/java/org/apache/gora/mapreduce/DataStoreMapReduceTestBase.java
    gora/trunk/gora-core/src/test/java/org/apache/gora/mapreduce/MapReduceTestUtils.java
    gora/trunk/gora-core/src/test/java/org/apache/gora/mapreduce/TestGoraInputFormat.java
    gora/trunk/gora-core/src/test/java/org/apache/gora/mapreduce/TestPersistentSerialization.java
    gora/trunk/gora-core/src/test/java/org/apache/gora/mock/persistency/MockPersistent.java
    gora/trunk/gora-core/src/test/java/org/apache/gora/persistency/TestListGenericArray.java
    gora/trunk/gora-core/src/test/java/org/apache/gora/persistency/impl/TestPersistentBase.java
    gora/trunk/gora-core/src/test/java/org/apache/gora/persistency/impl/TestStateManagerImpl.java
    gora/trunk/gora-core/src/test/java/org/apache/gora/store/DataStoreTestBase.java
    gora/trunk/gora-core/src/test/java/org/apache/gora/store/DataStoreTestUtil.java
    gora/trunk/gora-core/src/test/java/org/apache/gora/store/TestDataStoreFactory.java
    gora/trunk/gora-core/src/test/java/org/apache/gora/util/TestIOUtils.java
    gora/trunk/gora-dynamodb/src/main/java/org/apache/gora/dynamodb/store/DynamoDBStore.java
    gora/trunk/gora-hbase/pom.xml
    gora/trunk/gora-hbase/src/main/java/org/apache/gora/hbase/query/HBaseGetResult.java
    gora/trunk/gora-hbase/src/main/java/org/apache/gora/hbase/query/HBaseResult.java
    gora/trunk/gora-hbase/src/main/java/org/apache/gora/hbase/store/HBaseStore.java
    gora/trunk/gora-hbase/src/main/java/org/apache/gora/hbase/store/HBaseTableConnection.java
    gora/trunk/gora-hbase/src/main/java/org/apache/gora/hbase/util/HBaseByteInterface.java
    gora/trunk/gora-hbase/src/test/conf/gora-hbase-mapping.xml
    gora/trunk/gora-hbase/src/test/conf/hbase-site.xml
    gora/trunk/gora-hbase/src/test/java/org/apache/gora/hbase/GoraHBaseTestDriver.java
    gora/trunk/gora-hbase/src/test/java/org/apache/gora/hbase/mapreduce/TestHBaseStoreCountQuery.java
    gora/trunk/gora-hbase/src/test/java/org/apache/gora/hbase/store/TestHBaseStore.java
    gora/trunk/gora-hbase/src/test/java/org/apache/gora/hbase/util/HBaseClusterSingleton.java
    gora/trunk/gora-hbase/src/test/java/org/apache/gora/hbase/util/TestHBaseByteInterface.java
    gora/trunk/gora-solr/pom.xml
    gora/trunk/gora-solr/src/main/java/org/apache/gora/solr/store/SolrStore.java
    gora/trunk/gora-solr/src/test/conf/gora-solr-mapping.xml
    gora/trunk/gora-solr/src/test/conf/gora.properties
    gora/trunk/gora-solr/src/test/conf/solr/Employee/conf/schema.xml
    gora/trunk/gora-solr/src/test/conf/solr/WebPage/conf/schema.xml
    gora/trunk/gora-solr/src/test/java/org/apache/gora/solr/store/TestSolrStore.java
    gora/trunk/gora-tutorial/pom.xml
    gora/trunk/gora-tutorial/src/main/avro/metricdatum.json
    gora/trunk/gora-tutorial/src/main/avro/pageview.json
    gora/trunk/gora-tutorial/src/main/java/org/apache/gora/tutorial/log/LogAnalytics.java
    gora/trunk/gora-tutorial/src/main/java/org/apache/gora/tutorial/log/generated/MetricDatum.java
    gora/trunk/gora-tutorial/src/main/java/org/apache/gora/tutorial/log/generated/Pageview.java
    gora/trunk/pom.xml

Modified: gora/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/gora/trunk/CHANGES.txt?rev=1586888&r1=1586887&r2=1586888&view=diff
==============================================================================
--- gora/trunk/CHANGES.txt (original)
+++ gora/trunk/CHANGES.txt Sat Apr 12 19:21:53 2014
@@ -4,13 +4,29 @@
 
 Gora Change Log
 
+* GORA-106 Migrate Gora website documentation to Apache CMS (lewismc)
+
+* GORA-296 Improve 'Keyclass and nameclass match' logging in HBaseStore (rmarroquin via lewismc)
+
+* GORA-246 Upgrade to Avro 1.7.X in gora-hbase (Alparslan Avcı, rmarroquin, lewismc via lewismc)
+
+* GORA-154 delete() method is not implemented at CassandraStore, and always returns false or 0 (rmarroquin via Kazuomi Kashii)
+
+* GORA-204 Don't store empty arrays in CassandraClient#addGenericArray(), addStatefulHashMap() and CassandraStore#addOrUpdateField(rmarroquin via lewismc)
+
+* GORA-303 Upgrade to Avro 1.7.X in gora-solr (Talat UYARER)
+
+* GORA-253 Add Facebook, Linkedin, Google+, Twitter, etc plugins to website (lewismc)
+
+* GORA-244 Upgrade to Avro 1.7.X in gora-accumulo (Akber Choudhry via lewismc)
+
+* GORA-306 Ssn field is not nullable in Employee's Avro Schema (Talat UYARER via lewismc)
+
 * GORA-171 Implement Daily Rolling File Appender for localised Gora logging (lewismc)
 
 * GORA-119 implement a filter enabled scan in gora (ferdy, kturner, enis, Tien Nguyen Manh via lewismc)
 
-* GORA-290 StatefulHashMap removes the entry when put with same value (Alparslan Avci via hsaputra)
-
-* GORA-231 Provide better error handling in AccumuloStore.readMapping when file does not exist (Apostolos Giannakidis)
+* GORA-231 Provide better error handling in AccumuloStore.readMapping when file does not exist. (apgiannakidis via lewismc)
 
 * GORA-283 Specify field name for types not being considered in gora-cassandra (lewismc)
 

Modified: gora/trunk/bin/gora
URL: http://svn.apache.org/viewvc/gora/trunk/bin/gora?rev=1586888&r1=1586887&r2=1586888&view=diff
==============================================================================
--- gora/trunk/bin/gora (original)
+++ gora/trunk/bin/gora Sat Apr 12 19:21:53 2014
@@ -110,9 +110,9 @@ fi
 
 # figure out which class to run
 if [ "$COMMAND" = "goracompiler" ] ; then
-  MODULE=gora-core
+  MODULE=gora-compiler-cli
   CLASSPATH=$CLASSPATH:$GORA_HOME/$MODULE/target/classes/
-  CLASS=org.apache.gora.compiler.GoraCompiler
+  CLASS=org.apache.gora.compiler.cli.GoraCompilerCLI
 elif [ "$COMMAND" = "specificcompiler" ] ; then
   MODULE=gora-core
   CLASSPATH=$CLASSPATH:$GORA_HOME/$MODULE/target/classes/

Modified: gora/trunk/gora-accumulo/pom.xml
URL: http://svn.apache.org/viewvc/gora/trunk/gora-accumulo/pom.xml?rev=1586888&r1=1586887&r2=1586888&view=diff
==============================================================================
--- gora/trunk/gora-accumulo/pom.xml (original)
+++ gora/trunk/gora-accumulo/pom.xml Sat Apr 12 19:21:53 2014
@@ -1,22 +1,17 @@
 <?xml version="1.0" encoding="UTF-8"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-    <!--
-    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.
-    -->
-    
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<!-- 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. -->
+
 	<modelVersion>4.0.0</modelVersion>
 
 	<parent>
@@ -114,12 +109,12 @@
         <dependency>
            <groupId>org.apache.accumulo</groupId>
            <artifactId>accumulo-core</artifactId>
-           <version>1.4.0</version>
+           <version>1.5.1</version>
         </dependency>
 
         <!-- Hadoop Dependencies -->
         <dependency>
-            <groupId>org.apache.hadoop</groupId>
+            <groupId>org.apache.avro</groupId>
             <artifactId>avro</artifactId>
         </dependency>
 
@@ -145,6 +140,12 @@
             <artifactId>hadoop-test</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-simple</artifactId>
+            <scope>test</scope>
+        </dependency>
+           
     </dependencies>
 
 </project>

Modified: gora/trunk/gora-accumulo/src/main/java/org/apache/gora/accumulo/encoders/BinaryEncoder.java
URL: http://svn.apache.org/viewvc/gora/trunk/gora-accumulo/src/main/java/org/apache/gora/accumulo/encoders/BinaryEncoder.java?rev=1586888&r1=1586887&r2=1586888&view=diff
==============================================================================
--- gora/trunk/gora-accumulo/src/main/java/org/apache/gora/accumulo/encoders/BinaryEncoder.java (original)
+++ gora/trunk/gora-accumulo/src/main/java/org/apache/gora/accumulo/encoders/BinaryEncoder.java Sat Apr 12 19:21:53 2014
@@ -33,6 +33,7 @@ public class BinaryEncoder implements En
   
   public byte[] encodeShort(short s, byte ret[]) {
     try {
+      @SuppressWarnings("resource")
       DataOutputStream dos = new DataOutputStream(new FixedByteArrayOutputStream(ret));
       dos.writeShort(s);
       return ret;
@@ -57,6 +58,7 @@ public class BinaryEncoder implements En
   
   public byte[] encodeInt(int i, byte ret[]) {
     try {
+      @SuppressWarnings("resource")
       DataOutputStream dos = new DataOutputStream(new FixedByteArrayOutputStream(ret));
       dos.writeInt(i);
       return ret;
@@ -81,6 +83,7 @@ public class BinaryEncoder implements En
   
   public byte[] encodeLong(long l, byte ret[]) {
     try {
+      @SuppressWarnings("resource")
       DataOutputStream dos = new DataOutputStream(new FixedByteArrayOutputStream(ret));
       dos.writeLong(l);
       return ret;
@@ -106,6 +109,7 @@ public class BinaryEncoder implements En
   public byte[] encodeDouble(double d, byte[] ret) {
     try {
       long l = Double.doubleToRawLongBits(d);
+      @SuppressWarnings("resource")
       DataOutputStream dos = new DataOutputStream(new FixedByteArrayOutputStream(ret));
       dos.writeLong(l);
       return ret;
@@ -131,6 +135,7 @@ public class BinaryEncoder implements En
   public byte[] encodeFloat(float f, byte[] ret) {
     try {
       int i = Float.floatToRawIntBits(f);
+      @SuppressWarnings("resource")
       DataOutputStream dos = new DataOutputStream(new FixedByteArrayOutputStream(ret));
       dos.writeInt(i);
       return ret;
@@ -177,6 +182,7 @@ public class BinaryEncoder implements En
   
   public byte[] encodeBoolean(boolean b, byte[] ret) {
     try {
+      @SuppressWarnings("resource")
       DataOutputStream dos = new DataOutputStream(new FixedByteArrayOutputStream(ret));
       dos.writeBoolean(b);
       return ret;

Modified: gora/trunk/gora-accumulo/src/main/java/org/apache/gora/accumulo/query/AccumuloResult.java
URL: http://svn.apache.org/viewvc/gora/trunk/gora-accumulo/src/main/java/org/apache/gora/accumulo/query/AccumuloResult.java?rev=1586888&r1=1586887&r2=1586888&view=diff
==============================================================================
--- gora/trunk/gora-accumulo/src/main/java/org/apache/gora/accumulo/query/AccumuloResult.java (original)
+++ gora/trunk/gora-accumulo/src/main/java/org/apache/gora/accumulo/query/AccumuloResult.java Sat Apr 12 19:21:53 2014
@@ -75,7 +75,7 @@ public class AccumuloResult<K,T extends 
     
     Iterator<Entry<Key,Value>> nextRow = iterator.next();
     ByteSequence row = getDataStore().populate(nextRow, persistent);
-    key = (K) ((AccumuloStore) dataStore).fromBytes(getKeyClass(), row.toArray());
+    key = (K) ((AccumuloStore<K, T>) dataStore).fromBytes(getKeyClass(), row.toArray());
     
     return true;
   }

Modified: gora/trunk/gora-accumulo/src/main/java/org/apache/gora/accumulo/store/AccumuloStore.java
URL: http://svn.apache.org/viewvc/gora/trunk/gora-accumulo/src/main/java/org/apache/gora/accumulo/store/AccumuloStore.java?rev=1586888&r1=1586887&r2=1586888&view=diff
==============================================================================
--- gora/trunk/gora-accumulo/src/main/java/org/apache/gora/accumulo/store/AccumuloStore.java (original)
+++ gora/trunk/gora-accumulo/src/main/java/org/apache/gora/accumulo/store/AccumuloStore.java Sat Apr 12 19:21:53 2014
@@ -30,6 +30,7 @@ import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Properties;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 
 import javax.xml.parsers.DocumentBuilder;
 import javax.xml.parsers.DocumentBuilderFactory;
@@ -38,6 +39,7 @@ import org.apache.accumulo.core.Constant
 import org.apache.accumulo.core.client.AccumuloException;
 import org.apache.accumulo.core.client.AccumuloSecurityException;
 import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
 import org.apache.accumulo.core.client.Connector;
 import org.apache.accumulo.core.client.IsolatedScanner;
 import org.apache.accumulo.core.client.IteratorSetting;
@@ -54,6 +56,8 @@ import org.apache.accumulo.core.client.i
 import org.apache.accumulo.core.client.mock.MockConnector;
 import org.apache.accumulo.core.client.mock.MockInstance;
 import org.apache.accumulo.core.client.mock.MockTabletLocator;
+import org.apache.accumulo.core.client.security.tokens.AuthenticationToken;
+import org.apache.accumulo.core.client.security.tokens.PasswordToken;
 import org.apache.accumulo.core.data.ByteSequence;
 import org.apache.accumulo.core.data.Key;
 import org.apache.accumulo.core.data.KeyExtent;
@@ -64,27 +68,27 @@ import org.apache.accumulo.core.iterator
 import org.apache.accumulo.core.iterators.user.TimestampFilter;
 import org.apache.accumulo.core.master.state.tables.TableState;
 import org.apache.accumulo.core.security.ColumnVisibility;
-import org.apache.accumulo.core.security.thrift.AuthInfo;
+import org.apache.accumulo.core.security.thrift.TCredentials;
 import org.apache.accumulo.core.util.Pair;
 import org.apache.accumulo.core.util.TextUtil;
 import org.apache.accumulo.core.util.UtilWaitThread;
 import org.apache.avro.Schema;
 import org.apache.avro.Schema.Field;
-import org.apache.avro.generic.GenericArray;
+import org.apache.avro.Schema.Type;
+import org.apache.avro.generic.GenericData;
 import org.apache.avro.io.BinaryDecoder;
-import org.apache.avro.io.BinaryEncoder;
+import org.apache.avro.io.Decoder;
 import org.apache.avro.io.DecoderFactory;
+import org.apache.avro.io.EncoderFactory;
 import org.apache.avro.specific.SpecificDatumReader;
 import org.apache.avro.specific.SpecificDatumWriter;
 import org.apache.avro.util.Utf8;
+import org.apache.gora.accumulo.encoders.BinaryEncoder;
 import org.apache.gora.accumulo.encoders.Encoder;
 import org.apache.gora.accumulo.query.AccumuloQuery;
 import org.apache.gora.accumulo.query.AccumuloResult;
-import org.apache.gora.persistency.ListGenericArray;
-import org.apache.gora.persistency.State;
-import org.apache.gora.persistency.StateManager;
-import org.apache.gora.persistency.StatefulHashMap;
-import org.apache.gora.persistency.StatefulMap;
+import org.apache.gora.persistency.impl.DirtyListWrapper;
+import org.apache.gora.persistency.impl.DirtyMapWrapper;
 import org.apache.gora.persistency.impl.PersistentBase;
 import org.apache.gora.query.PartitionQuery;
 import org.apache.gora.query.Query;
@@ -105,7 +109,7 @@ import org.w3c.dom.NodeList;
  * 
  */
 public class AccumuloStore<K,T extends PersistentBase> extends DataStoreBase<K,T> {
-  
+
   protected static final String MOCK_PROPERTY = "accumulo.mock";
   protected static final String INSTANCE_NAME_PROPERTY = "accumulo.instance";
   protected static final String ZOOKEEPERS_NAME_PROPERTY = "accumulo.zookeepers";
@@ -116,36 +120,71 @@ public class AccumuloStore<K,T extends P
   private Connector conn;
   private BatchWriter batchWriter;
   private AccumuloMapping mapping;
-  private AuthInfo authInfo;
+  private TCredentials credentials;
   private Encoder encoder;
-  
+
   public static final Logger LOG = LoggerFactory.getLogger(AccumuloStore.class);
-  
-  public Object fromBytes(Schema schema, byte data[]) {
-    return fromBytes(encoder, schema, data);
+
+  public Object fromBytes(Schema schema, byte data[]) throws GoraException {
+    Schema fromSchema = null;
+    if (schema.getType() == Type.UNION) {
+      try {
+        Decoder decoder = DecoderFactory.get().binaryDecoder(data, null);
+        int unionIndex = decoder.readIndex();
+        List<Schema> possibleTypes = schema.getTypes();
+        fromSchema = possibleTypes.get(unionIndex);
+        Schema effectiveSchema = possibleTypes.get(unionIndex);
+        if (effectiveSchema.getType() == Type.NULL) {
+          decoder.readNull();
+          return null;
+        } else {
+          data = decoder.readBytes(null).array();
+        }
+      } catch (IOException e) {
+        e.printStackTrace();
+        throw new GoraException("Error decoding union type: ", e);
+      }
+    } else {
+      fromSchema = schema;
+    }
+    return fromBytes(encoder, fromSchema, data);
   }
 
   public static Object fromBytes(Encoder encoder, Schema schema, byte data[]) {
     switch (schema.getType()) {
-      case BOOLEAN:
-        return encoder.decodeBoolean(data);
-      case DOUBLE:
-        return encoder.decodeDouble(data);
-      case FLOAT:
-        return encoder.decodeFloat(data);
-      case INT:
-        return encoder.decodeInt(data);
-      case LONG:
-        return encoder.decodeLong(data);
-      case STRING:
-        return new Utf8(data);
-      case BYTES:
-        return ByteBuffer.wrap(data);
-      case ENUM:
-        return AvroUtils.getEnumValue(schema, encoder.decodeInt(data));
+    case BOOLEAN:
+      return encoder.decodeBoolean(data);
+    case DOUBLE:
+      return encoder.decodeDouble(data);
+    case FLOAT:
+      return encoder.decodeFloat(data);
+    case INT:
+      return encoder.decodeInt(data);
+    case LONG:
+      return encoder.decodeLong(data);
+    case STRING:
+      return new Utf8(data);
+    case BYTES:
+      return ByteBuffer.wrap(data);
+    case ENUM:
+      return AvroUtils.getEnumValue(schema, encoder.decodeInt(data));
+    case ARRAY:
+      break;
+    case FIXED:
+      break;
+    case MAP:
+      break;
+    case NULL:
+      break;
+    case RECORD:
+      break;
+    case UNION:
+      break;
+    default:
+      break;
     }
     throw new IllegalArgumentException("Unknown type " + schema.getType());
-    
+
   }
 
   public K fromBytes(Class<K> clazz, byte[] val) {
@@ -174,7 +213,7 @@ public class AccumuloStore<K,T extends P
       } else if (clazz.equals(Utf8.class)) {
         return (K) new Utf8(val);
       }
-      
+
       throw new IllegalArgumentException("Unknown type " + clazz.getName());
     } catch (IOException ioe) {
       throw new RuntimeException(ioe);
@@ -190,17 +229,67 @@ public class AccumuloStore<K,T extends P
     return b;
   }
 
+  public byte[] toBytes(Schema toSchema, Object o) {
+    if (toSchema != null && toSchema.getType() == Type.UNION) {
+      ByteArrayOutputStream baos = new ByteArrayOutputStream();
+      org.apache.avro.io.BinaryEncoder avroEncoder = EncoderFactory.get().binaryEncoder(baos, null);
+      int unionIndex = 0;
+      try {
+        if (o == null) {
+          unionIndex = firstNullSchemaTypeIndex(toSchema);
+          avroEncoder.writeIndex(unionIndex);
+          avroEncoder.writeNull();
+        } else {
+          unionIndex = firstNotNullSchemaTypeIndex(toSchema);
+          avroEncoder.writeIndex(unionIndex);
+          avroEncoder.writeBytes(toBytes(o));
+        }
+        avroEncoder.flush();
+        return baos.toByteArray();
+      } catch (IOException e) {
+        e.printStackTrace();
+        return toBytes(o);
+      }
+    } else {     
+      return toBytes(o);
+    }
+  }
+
+  private int firstNullSchemaTypeIndex(Schema toSchema) {
+    List<Schema> possibleTypes = toSchema.getTypes();
+    int unionIndex = 0;
+    for (int i = 0; i < possibleTypes.size(); i++ ) {
+      Type pType = possibleTypes.get(i).getType();
+      if (pType == Type.NULL) { // FIXME HUGE kludge to pass tests
+        unionIndex = i; break;
+      }
+    }
+    return unionIndex;
+  }
+
+  private int firstNotNullSchemaTypeIndex(Schema toSchema) {
+    List<Schema> possibleTypes = toSchema.getTypes();
+    int unionIndex = 0;
+    for (int i = 0; i < possibleTypes.size(); i++ ) {
+      Type pType = possibleTypes.get(i).getType();
+      if (pType != Type.NULL) { // FIXME HUGE kludge to pass tests
+        unionIndex = i; break;
+      }
+    }
+    return unionIndex;
+  }
+
   public byte[] toBytes(Object o) {
     return toBytes(encoder, o);
   }
-  
+
   public static byte[] toBytes(Encoder encoder, Object o) {
-    
+
     try {
       if (o instanceof String) {
         return ((String) o).getBytes("UTF-8");
       } else if (o instanceof Utf8) {
-        return copyIfNeeded(((Utf8) o).getBytes(), 0, ((Utf8) o).getLength());
+        return copyIfNeeded(((Utf8) o).getBytes(), 0, ((Utf8) o).getByteLength());
       } else if (o instanceof ByteBuffer) {
         return copyIfNeeded(((ByteBuffer) o).array(), ((ByteBuffer) o).arrayOffset() + ((ByteBuffer) o).position(), ((ByteBuffer) o).remaining());
       } else if (o instanceof Long) {
@@ -218,19 +307,23 @@ public class AccumuloStore<K,T extends P
       } else if (o instanceof Double) {
         return encoder.encodeDouble((Double) o);
       } else if (o instanceof Enum) {
-        return encoder.encodeInt(((Enum) o).ordinal());
+        return encoder.encodeInt(((Enum<?>) o).ordinal());
       }
     } catch (IOException ioe) {
       throw new RuntimeException(ioe);
     }
-    
+
     throw new IllegalArgumentException("Uknown type " + o.getClass().getName());
   }
 
   private BatchWriter getBatchWriter() throws IOException {
     if (batchWriter == null)
       try {
-        batchWriter = conn.createBatchWriter(mapping.tableName, 10000000, 60000l, 4);
+        BatchWriterConfig batchWriterConfig = new BatchWriterConfig();
+        batchWriterConfig.setMaxMemory(10000000);
+        batchWriterConfig.setMaxLatency(60000l, TimeUnit.MILLISECONDS);
+        batchWriterConfig.setMaxWriteThreads(4);
+        batchWriter = conn.createBatchWriter(mapping.tableName, batchWriterConfig);
       } catch (TableNotFoundException e) {
         throw new IOException(e);
       }
@@ -241,16 +334,16 @@ public class AccumuloStore<K,T extends P
   public void initialize(Class<K> keyClass, Class<T> persistentClass, Properties properties) {
     try{
       super.initialize(keyClass, persistentClass, properties);
-  
+
       String mock = DataStoreFactory.findProperty(properties, this, MOCK_PROPERTY, null);
       String mappingFile = DataStoreFactory.getMappingFile(properties, this, DEFAULT_MAPPING_FILE);
       String user = DataStoreFactory.findProperty(properties, this, USERNAME_PROPERTY, null);
       String password = DataStoreFactory.findProperty(properties, this, PASSWORD_PROPERTY, null);
-      
+
       mapping = readMapping(mappingFile);
-  
+
       if (mapping.encoder == null || mapping.encoder.equals("")) {
-        encoder = new org.apache.gora.accumulo.encoders.BinaryEncoder();
+        encoder = new BinaryEncoder();
       } else {
         try {
           encoder = (Encoder) getClass().getClassLoader().loadClass(mapping.encoder).newInstance();
@@ -262,17 +355,23 @@ public class AccumuloStore<K,T extends P
           throw new IOException(e);
         }
       }
-  
+
       try {
+        AuthenticationToken token =  new PasswordToken(password);
         if (mock == null || !mock.equals("true")) {
           String instance = DataStoreFactory.findProperty(properties, this, INSTANCE_NAME_PROPERTY, null);
           String zookeepers = DataStoreFactory.findProperty(properties, this, ZOOKEEPERS_NAME_PROPERTY, null);
-          conn = new ZooKeeperInstance(instance, zookeepers).getConnector(user, password);
-          authInfo = new AuthInfo(user, ByteBuffer.wrap(password.getBytes()), conn.getInstance().getInstanceID());
+          credentials = new TCredentials(user, 
+              "org.apache.accumulo.core.client.security.tokens.PasswordToken", 
+              ByteBuffer.wrap(password.getBytes()), instance);
+          conn = new ZooKeeperInstance(instance, zookeepers).getConnector(user, token);
         } else {
-          conn = new MockInstance().getConnector(user, password);
+          conn = new MockInstance().getConnector(user, new PasswordToken(password));
+          credentials = new TCredentials(user, 
+              "org.apache.accumulo.core.client.security.tokens.PasswordToken", 
+              ByteBuffer.wrap(password.getBytes()), conn.getInstance().getInstanceID());
         }
-  
+
         if (autoCreateSchema)
           createSchema();
       } catch (AccumuloException e) {
@@ -284,27 +383,27 @@ public class AccumuloStore<K,T extends P
       LOG.error(e.getMessage(), e);
     }
   }
-  
+
   protected AccumuloMapping readMapping(String filename) throws IOException {
     try {
-      
+
       AccumuloMapping mapping = new AccumuloMapping();
 
       DocumentBuilder db = DocumentBuilderFactory.newInstance().newDocumentBuilder();
       Document dom = db.parse(getClass().getClassLoader().getResourceAsStream(filename));
-      
+
       Element root = dom.getDocumentElement();
-      
+
       NodeList nl = root.getElementsByTagName("class");
       for (int i = 0; i < nl.getLength(); i++) {
-        
+
         Element classElement = (Element) nl.item(i);
         if (classElement.getAttribute("keyClass").equals(keyClass.getCanonicalName())
             && classElement.getAttribute("name").equals(persistentClass.getCanonicalName())) {
 
           mapping.tableName = getSchemaName(classElement.getAttribute("table"), persistentClass);
           mapping.encoder = classElement.getAttribute("encoder");
-          
+
           NodeList fields = classElement.getElementsByTagName("field");
           for (int j = 0; j < fields.getLength(); j++) {
             Element fieldElement = (Element) fields.item(j);
@@ -324,9 +423,9 @@ public class AccumuloStore<K,T extends P
       }
 
       if (mapping.tableName == null) {
-        throw new GoraException("Please define the gora to accumulo mapping in " + filename + " for " + persistentClass.getCanonicalName());
+        throw new GoraException("Please define the accumulo 'table' name mapping in " + filename + " for " + persistentClass.getCanonicalName());
       }
-      
+
       nl = root.getElementsByTagName("table");
       for (int i = 0; i < nl.getLength(); i++) {
         Element tableElement = (Element) nl.item(i);
@@ -347,12 +446,12 @@ public class AccumuloStore<K,T extends P
     }
 
   }
-  
+
   @Override
   public String getSchemaName() {
     return mapping.tableName;
   }
-  
+
   @Override
   public void createSchema() {
     try {
@@ -394,20 +493,30 @@ public class AccumuloStore<K,T extends P
 
   public ByteSequence populate(Iterator<Entry<Key,Value>> iter, T persistent) throws IOException {
     ByteSequence row = null;
-    
-    Map currentMap = null;
-    ArrayList currentArray = null;
+
+    Map<Utf8, Object> currentMap = null;
+    List currentArray = null;
     Text currentFam = null;
     int currentPos = 0;
     Schema currentSchema = null;
     Field currentField = null;
 
+    BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(new byte[0], null);
+
     while (iter.hasNext()) {
       Entry<Key,Value> entry = iter.next();
-      
+
+      if (row == null) {
+        row = entry.getKey().getRowData();
+      }
+      byte[] val = entry.getValue().get();
+
+      Field field = fieldMap.get(getFieldName(entry));
+
       if (currentMap != null) {
         if (currentFam.equals(entry.getKey().getColumnFamily())) {
-          currentMap.put(new Utf8(entry.getKey().getColumnQualifierData().toArray()), fromBytes(currentSchema, entry.getValue().get()));
+          currentMap.put(new Utf8(entry.getKey().getColumnQualifierData().toArray()), 
+              fromBytes(currentSchema, entry.getValue().get()));
           continue;
         } else {
           persistent.put(currentPos, currentMap);
@@ -418,57 +527,69 @@ public class AccumuloStore<K,T extends P
           currentArray.add(fromBytes(currentSchema, entry.getValue().get()));
           continue;
         } else {
-          persistent.put(currentPos, new ListGenericArray<T>(currentField.schema(), currentArray));
+          persistent.put(currentPos, new GenericData.Array<T>(currentField.schema(), currentArray));
           currentArray = null;
         }
       }
 
-      if (row == null)
-        row = entry.getKey().getRowData();
-      
-      String fieldName = mapping.columnMap.get(new Pair<Text,Text>(entry.getKey().getColumnFamily(), entry.getKey().getColumnQualifier()));
-      if (fieldName == null)
-        fieldName = mapping.columnMap.get(new Pair<Text,Text>(entry.getKey().getColumnFamily(), null));
-
-      Field field = fieldMap.get(fieldName);
-
       switch (field.schema().getType()) {
-        case MAP:
-          currentMap = new StatefulHashMap();
+      case MAP:  // first entry only. Next are handled above on the next loop
+        currentMap = new DirtyMapWrapper<Utf8, Object>(new HashMap<Utf8, Object>());
+        currentPos = field.pos();
+        currentFam = entry.getKey().getColumnFamily();
+        currentSchema = field.schema().getValueType();
+
+        currentMap.put(new Utf8(entry.getKey().getColumnQualifierData().toArray()), 
+            fromBytes(currentSchema, entry.getValue().get()));
+        break;
+      case ARRAY:
+        currentArray = new DirtyListWrapper<Object>(new ArrayList<Object>());
+        currentPos = field.pos();
+        currentFam = entry.getKey().getColumnFamily();
+        currentSchema = field.schema().getElementType();
+        currentField = field;
+
+        currentArray.add(fromBytes(currentSchema, entry.getValue().get()));
+
+        break;
+      case UNION:// default value of null acts like union with null
+        Schema effectiveSchema = field.schema().getTypes()
+        .get(firstNotNullSchemaTypeIndex(field.schema()));
+        // map and array were coded without union index so need to be read the same way
+        if (effectiveSchema.getType() == Type.ARRAY) {
+          currentArray = new DirtyListWrapper<Object>(new ArrayList<Object>());
           currentPos = field.pos();
           currentFam = entry.getKey().getColumnFamily();
-          currentSchema = field.schema().getValueType();
-          
-          currentMap.put(new Utf8(entry.getKey().getColumnQualifierData().toArray()), fromBytes(currentSchema, entry.getValue().get()));
+          currentSchema = field.schema().getElementType();
+          currentField = field;
 
+          currentArray.add(fromBytes(currentSchema, entry.getValue().get()));
           break;
-        case ARRAY:
-          currentArray = new ArrayList();
+        }
+        else if (effectiveSchema.getType() == Type.MAP) {
+          currentMap = new DirtyMapWrapper<Utf8, Object>(new HashMap<Utf8, Object>());
           currentPos = field.pos();
           currentFam = entry.getKey().getColumnFamily();
-          currentSchema = field.schema().getElementType();
-          currentField = field;
-          
-          currentArray.add(fromBytes(currentSchema, entry.getValue().get()));
+          currentSchema = effectiveSchema.getValueType();
 
+          currentMap.put(new Utf8(entry.getKey().getColumnQualifierData().toArray()), 
+              fromBytes(currentSchema, entry.getValue().get()));
           break;
-        case RECORD:
-        case UNION:
-          SpecificDatumReader reader = new SpecificDatumReader(field.schema());
-          byte[] val = entry.getValue().get();
-          // TODO reuse decoder
-          BinaryDecoder decoder = DecoderFactory.defaultFactory().createBinaryDecoder(val, null);
-          persistent.put(field.pos(), reader.read(null, decoder));
-          break;
-        default:
-          persistent.put(field.pos(), fromBytes(field.schema(), entry.getValue().get()));
+        }
+        // continue like a regular top-level union
+      case RECORD:
+        SpecificDatumReader<?> reader = new SpecificDatumReader<Schema>(field.schema());
+        persistent.put(field.pos(), reader.read(null, DecoderFactory.get().binaryDecoder(val, decoder)));
+        break;
+      default:
+        persistent.put(field.pos(), fromBytes(field.schema(), entry.getValue().get()));
       }
     }
-    
+
     if (currentMap != null) {
       persistent.put(currentPos, currentMap);
     } else if (currentArray != null) {
-      persistent.put(currentPos, new ListGenericArray<T>(currentField.schema(), currentArray));
+      persistent.put(currentPos, new GenericData.Array<T>(currentField.schema(), currentArray));
     }
 
     persistent.clearDirty();
@@ -476,14 +597,32 @@ public class AccumuloStore<K,T extends P
     return row;
   }
 
+  /**
+   * Retrieve field name from entry.
+   * @param entry The Key-Value entry
+   * @return String The field name
+   */
+  private String getFieldName(Entry<Key, Value> entry) {
+    String fieldName = mapping.columnMap.get(new Pair<Text,Text>(entry.getKey().getColumnFamily(), 
+        entry.getKey().getColumnQualifier()));
+    if (fieldName == null) {
+      fieldName = mapping.columnMap.get(new Pair<Text,Text>(entry.getKey().getColumnFamily(), null));
+    }
+    return fieldName;
+  }
+
   private void setFetchColumns(Scanner scanner, String fields[]) {
     fields = getFieldsToQuery(fields);
     for (String field : fields) {
       Pair<Text,Text> col = mapping.fieldMap.get(field);
-      if (col.getSecond() == null) {
-        scanner.fetchColumnFamily(col.getFirst());
+      if (col != null) {
+        if (col.getSecond() == null) {
+          scanner.fetchColumnFamily(col.getFirst());
+        } else {
+          scanner.fetchColumn(col.getFirst(), col.getSecond());
+        }
       } else {
-        scanner.fetchColumn(col.getFirst(), col.getSecond());
+        LOG.error("Mapping not found for field: " + field);
       }
     }
   }
@@ -494,10 +633,10 @@ public class AccumuloStore<K,T extends P
       // TODO make isolated scanner optional?
       Scanner scanner = new IsolatedScanner(conn.createScanner(mapping.tableName, Constants.NO_AUTHS));
       Range rowRange = new Range(new Text(toBytes(key)));
-      
+
       scanner.setRange(rowRange);
       setFetchColumns(scanner, fields);
-      
+
       T persistent = newPersistent();
       ByteSequence row = populate(scanner.iterator(), persistent);
       if (row == null)
@@ -511,90 +650,67 @@ public class AccumuloStore<K,T extends P
       return null;
     }
   }
-  
+
   @Override
   public void put(K key, T val) {
 
     try{
       Mutation m = new Mutation(new Text(toBytes(key)));
-      
+
       Schema schema = val.getSchema();
-      StateManager stateManager = val.getStateManager();
-      
-      Iterator<Field> iter = schema.getFields().iterator();
-      
+      List<Field> fields = schema.getFields();
       int count = 0;
-      for (int i = 0; iter.hasNext(); i++) {
-        Field field = iter.next();
-        if (!stateManager.isDirty(val, i)) {
+
+      for (int i = 1; i < fields.size(); i++) {
+        if (!val.isDirty(i)) {
           continue;
         }
-        
-        Object o = val.get(i);
+        Field field = fields.get(i);
+
+        Object o = val.get(field.pos());       
+
         Pair<Text,Text> col = mapping.fieldMap.get(field.name());
 
         if (col == null) {
           throw new GoraException("Please define the gora to accumulo mapping for field " + field.name());
         }
 
-  
         switch (field.schema().getType()) {
-          case MAP:
-            if (o instanceof StatefulMap) {
-              StatefulMap map = (StatefulMap) o;
-              Set<?> es = map.states().entrySet();
-              for (Object entry : es) {
-                Object mapKey = ((Entry) entry).getKey();
-                State state = (State) ((Entry) entry).getValue();
-  
-                switch (state) {
-                  case NEW:
-                  case DIRTY:
-                    m.put(col.getFirst(), new Text(toBytes(mapKey)), new Value(toBytes(map.get(mapKey))));
-                    count++;
-                    break;
-                  case DELETED:
-                    m.putDelete(col.getFirst(), new Text(toBytes(mapKey)));
-                    count++;
-                    break;
-                }
-                
-              }
-            } else {
-              Map map = (Map) o;
-              Set<?> es = map.entrySet();
-              for (Object entry : es) {
-                Object mapKey = ((Entry) entry).getKey();
-                Object mapVal = ((Entry) entry).getValue();
-                m.put(col.getFirst(), new Text(toBytes(mapKey)), new Value(toBytes(mapVal)));
-                count++;
-              }
-            }
-            break;
-          case ARRAY:
-            GenericArray array = (GenericArray) o;
-            int j = 0;
-            for (Object item : array) {
-              m.put(col.getFirst(), new Text(toBytes(j++)), new Value(toBytes(item)));
-              count++;
-            }
+        case MAP:
+          count = putMap(m, count, field.schema().getValueType(), o, col);
+          break;
+        case ARRAY:
+          count = putArray(m, count, o, col);
+          break;
+        case UNION: // default value of null acts like union with null
+          Schema effectiveSchema = field.schema().getTypes()
+          .get(firstNotNullSchemaTypeIndex(field.schema()));
+          // map and array need to compute qualifier
+          if (effectiveSchema.getType() == Type.ARRAY) {
+            count = putArray(m, count, o, col);
             break;
-          case RECORD:
-          case UNION:
-            SpecificDatumWriter writer = new SpecificDatumWriter(field.schema());
-            ByteArrayOutputStream os = new ByteArrayOutputStream();
-            BinaryEncoder encoder = new BinaryEncoder(os);
-            writer.write(o, encoder);
-            encoder.flush();
-            m.put(col.getFirst(), col.getSecond(), new Value(os.toByteArray()));
+          }
+          else if (effectiveSchema.getType() == Type.MAP) {
+            count = putMap(m, count, effectiveSchema.getValueType(), o, col);
             break;
-          default:
-            m.put(col.getFirst(), col.getSecond(), new Value(toBytes(o)));
-            count++;
+          }
+          // continue like a regular top-level union
+        case RECORD:
+          SpecificDatumWriter<Object> writer = new SpecificDatumWriter<Object>(field.schema());
+          ByteArrayOutputStream os = new ByteArrayOutputStream();
+          org.apache.avro.io.BinaryEncoder encoder = EncoderFactory.get().binaryEncoder(os, null);
+          writer.write(o, encoder);
+          encoder.flush();
+          m.put(col.getFirst(), col.getSecond(), new Value(os.toByteArray()));
+          count++;
+          break;
+        default:
+          m.put(col.getFirst(), col.getSecond(), new Value(toBytes(o)));
+          count++;
         }
-  
+
       }
-      
+
       if (count > 0)
         try {
           getBatchWriter().addMutation(m);
@@ -605,7 +721,58 @@ public class AccumuloStore<K,T extends P
       LOG.error(e.getMessage(), e);
     }
   }
-  
+
+  private int putMap(Mutation m, int count, Schema valueType, Object o, Pair<Text, Text> col) throws GoraException {
+
+    // First of all we delete map field on accumulo store
+    Text rowKey = new Text(m.getRow());
+    Query<K, T> query = newQuery();
+    query.setFields(col.getFirst().toString());
+    query.setStartKey((K)rowKey.toString());
+    query.setEndKey((K)rowKey.toString());
+    deleteByQuery(query);
+    flush();
+    if (o == null){
+      return 0;
+    }
+    
+    Set<?> es = ((Map<?, ?>)o).entrySet();
+    for (Object entry : es) {
+      Object mapKey = ((Entry<?, ?>) entry).getKey();
+      Object mapVal = ((Entry<?, ?>) entry).getValue();                  
+      if ((o instanceof DirtyMapWrapper && ((DirtyMapWrapper<?, ?>)o).isDirty())
+          || !(o instanceof DirtyMapWrapper)) { //mapVal instanceof Dirtyable && ((Dirtyable)mapVal).isDirty()) {
+        m.put(col.getFirst(), new Text(toBytes(mapKey)), new Value(toBytes(valueType, mapVal)));
+        count++;
+      }
+      // TODO map value deletion
+    }
+    return count;
+  }
+
+  private int putArray(Mutation m, int count, Object o, Pair<Text, Text> col) {
+
+    // First of all we delete array field on accumulo store
+    Text rowKey = new Text(m.getRow());
+    Query<K, T> query = newQuery();
+    query.setFields(col.getFirst().toString());
+    query.setStartKey((K)rowKey.toString());
+    query.setEndKey((K)rowKey.toString());
+    deleteByQuery(query);
+    flush();
+    if (o == null){
+      return 0;
+    }
+    
+    List<?> array = (List<?>) o;  // both GenericArray and DirtyListWrapper
+    int j = 0;
+    for (Object item : array) {
+      m.put(col.getFirst(), new Text(toBytes(j++)), new Value(toBytes(item)));
+      count++;
+    }
+    return count;
+  }
+
   @Override
   public boolean delete(K key) {
     Query<K,T> q = newQuery();
@@ -620,7 +787,7 @@ public class AccumuloStore<K,T extends P
       // add iterator that drops values on the server side
       scanner.addScanIterator(new IteratorSetting(Integer.MAX_VALUE, SortedKeyIterator.class));
       RowIterator iterator = new RowIterator(scanner.iterator());
-      
+
       long count = 0;
 
       while (iterator.hasNext()) {
@@ -637,7 +804,7 @@ public class AccumuloStore<K,T extends P
         getBatchWriter().addMutation(m);
         count++;
       }
-      
+
       return count;
     } catch (TableNotFoundException e) {
       // TODO return 0?
@@ -655,34 +822,34 @@ public class AccumuloStore<K,T extends P
   private Range createRange(Query<K,T> query) {
     Text startRow = null;
     Text endRow = null;
-    
+
     if (query.getStartKey() != null)
       startRow = new Text(toBytes(query.getStartKey()));
-    
+
     if (query.getEndKey() != null)
       endRow = new Text(toBytes(query.getEndKey()));
-    
+
     return new Range(startRow, true, endRow, true);
-    
+
   }
-  
+
   private Scanner createScanner(Query<K,T> query) throws TableNotFoundException {
     // TODO make isolated scanner optional?
     Scanner scanner = new IsolatedScanner(conn.createScanner(mapping.tableName, Constants.NO_AUTHS));
     setFetchColumns(scanner, query.getFields());
-    
+
     scanner.setRange(createRange(query));
-    
+
     if (query.getStartTime() != -1 || query.getEndTime() != -1) {
       IteratorSetting is = new IteratorSetting(30, TimestampFilter.class);
       if (query.getStartTime() != -1)
         TimestampFilter.setStart(is, query.getStartTime(), true);
       if (query.getEndTime() != -1)
         TimestampFilter.setEnd(is, query.getEndTime(), true);
-      
+
       scanner.addScanIterator(is);
     }
-    
+
     return scanner;
   }
 
@@ -697,7 +864,7 @@ public class AccumuloStore<K,T extends P
       return null;
     } 
   }
-  
+
   @Override
   public Query<K,T> newQuery() {
     return new AccumuloQuery<K,T>(this);
@@ -706,14 +873,14 @@ public class AccumuloStore<K,T extends P
   Text pad(Text key, int bytes) {
     if (key.getLength() < bytes)
       key = new Text(key);
-    
+
     while (key.getLength() < bytes) {
       key.append(new byte[] {0}, 0, 1);
     }
-    
+
     return key;
   }
-  
+
   @Override
   public List<PartitionQuery<K,T>> getPartitions(Query<K,T> query) throws IOException {
     try {
@@ -721,12 +888,12 @@ public class AccumuloStore<K,T extends P
       if (conn instanceof MockConnector)
         tl = new MockTabletLocator();
       else
-        tl = TabletLocator.getInstance(conn.getInstance(), authInfo, new Text(Tables.getTableId(conn.getInstance(), mapping.tableName)));
-      
+        tl = TabletLocator.getInstance(conn.getInstance(), new Text(Tables.getTableId(conn.getInstance(), mapping.tableName)));
+
       Map<String,Map<KeyExtent,List<Range>>> binnedRanges = new HashMap<String,Map<KeyExtent,List<Range>>>();
-      
+
       tl.invalidateCache();
-      while (tl.binRanges(Collections.singletonList(createRange(query)), binnedRanges).size() > 0) {
+      while (tl.binRanges(Collections.singletonList(createRange(query)), binnedRanges, credentials).size() > 0) {
         // TODO log?
         if (!Tables.exists(conn.getInstance(), Tables.getTableId(conn.getInstance(), mapping.tableName)))
           throw new TableDeletedException(Tables.getTableId(conn.getInstance(), mapping.tableName));
@@ -735,19 +902,19 @@ public class AccumuloStore<K,T extends P
         UtilWaitThread.sleep(100);
         tl.invalidateCache();
       }
-      
+
       List<PartitionQuery<K,T>> ret = new ArrayList<PartitionQuery<K,T>>();
-      
+
       Text startRow = null;
       Text endRow = null;
       if (query.getStartKey() != null)
         startRow = new Text(toBytes(query.getStartKey()));
       if (query.getEndKey() != null)
         endRow = new Text(toBytes(query.getEndKey()));
-     
+
       //hadoop expects hostnames, accumulo keeps track of IPs... so need to convert
       HashMap<String,String> hostNameCache = new HashMap<String,String>();
- 
+
       for (Entry<String,Map<KeyExtent,List<Range>>> entry : binnedRanges.entrySet()) {
         String ip = entry.getKey().split(":", 2)[0];
         String location = hostNameCache.get(ip);
@@ -759,7 +926,7 @@ public class AccumuloStore<K,T extends P
 
         Map<KeyExtent,List<Range>> tablets = entry.getValue();
         for (KeyExtent ke : tablets.keySet()) {
-          
+
           K startKey = null;
           if (startRow == null || !ke.contains(startRow)) {
             if (ke.getPrevEndRow() != null) {
@@ -768,7 +935,7 @@ public class AccumuloStore<K,T extends P
           } else {
             startKey = fromBytes(getKeyClass(), TextUtil.getBytes(startRow));
           }
-          
+
           K endKey = null;
           if (endRow == null || !ke.contains(endRow)) {
             if (ke.getEndRow() != null)
@@ -776,13 +943,13 @@ public class AccumuloStore<K,T extends P
           } else {
             endKey = fromBytes(getKeyClass(), TextUtil.getBytes(endRow));
           }
-          
-          PartitionQueryImpl pqi = new PartitionQueryImpl<K,T>(query, startKey, endKey, new String[] {location});
+
+          PartitionQueryImpl<K, T> pqi = new PartitionQueryImpl<K,T>(query, startKey, endKey, new String[] {location});
           pqi.setConf(getConf());
           ret.add(pqi);
         }
       }
-      
+
       return ret;
     } catch (TableNotFoundException e) {
       throw new IOException(e);
@@ -791,11 +958,11 @@ public class AccumuloStore<K,T extends P
     } catch (AccumuloSecurityException e) {
       throw new IOException(e);
     }
-    
+
   }
-  
+
   static <K> K lastPossibleKey(Encoder encoder, Class<K> clazz, byte[] er) {
-    
+
     if (clazz.equals(Byte.TYPE) || clazz.equals(Byte.class)) {
       throw new UnsupportedOperationException();
     } else if (clazz.equals(Boolean.TYPE) || clazz.equals(Boolean.class)) {
@@ -815,19 +982,20 @@ public class AccumuloStore<K,T extends P
     } else if (clazz.equals(Utf8.class)) {
       return fromBytes(encoder, clazz, er);
     }
-    
+
     throw new IllegalArgumentException("Unknown type " + clazz.getName());
   }
 
 
-  
+
   /**
    * @param keyClass
    * @param bytes
    * @return
    */
+  @SuppressWarnings("unchecked")
   static <K> K followingKey(Encoder encoder, Class<K> clazz, byte[] per) {
-    
+
     if (clazz.equals(Byte.TYPE) || clazz.equals(Byte.class)) {
       return (K) Byte.valueOf(encoder.followingKey(1, per)[0]);
     } else if (clazz.equals(Boolean.TYPE) || clazz.equals(Boolean.class)) {

Modified: gora/trunk/gora-accumulo/src/test/resources/gora-accumulo-mapping.xml
URL: http://svn.apache.org/viewvc/gora/trunk/gora-accumulo/src/test/resources/gora-accumulo-mapping.xml?rev=1586888&r1=1586887&r2=1586888&view=diff
==============================================================================
--- gora/trunk/gora-accumulo/src/test/resources/gora-accumulo-mapping.xml (original)
+++ gora/trunk/gora-accumulo/src/test/resources/gora-accumulo-mapping.xml Sat Apr 12 19:21:53 2014
@@ -47,6 +47,7 @@
     <field name="content" family="content" qualifier="c"/>
     <field name="parsedContent" family="parsedContent"/>
     <field name="outlinks" family="outlinks"/>
+    <field name="headers" family="headers"/>
     <field name="metadata" family="common" qualifier="metadata"/>
   </class>
 

Modified: gora/trunk/gora-accumulo/src/test/resources/gora.properties
URL: http://svn.apache.org/viewvc/gora/trunk/gora-accumulo/src/test/resources/gora.properties?rev=1586888&r1=1586887&r2=1586888&view=diff
==============================================================================
--- gora/trunk/gora-accumulo/src/test/resources/gora.properties (original)
+++ gora/trunk/gora-accumulo/src/test/resources/gora.properties Sat Apr 12 19:21:53 2014
@@ -18,4 +18,4 @@ gora.datastore.accumulo.mock=true
 gora.datastore.accumulo.instance=a14
 gora.datastore.accumulo.zookeepers=localhost
 gora.datastore.accumulo.user=root
-gora.datastore.accumulo.password=secret
\ No newline at end of file
+gora.datastore.accumulo.password=
\ No newline at end of file

Modified: gora/trunk/gora-cassandra/pom.xml
URL: http://svn.apache.org/viewvc/gora/trunk/gora-cassandra/pom.xml?rev=1586888&r1=1586887&r2=1586888&view=diff
==============================================================================
--- gora/trunk/gora-cassandra/pom.xml (original)
+++ gora/trunk/gora-cassandra/pom.xml Sat Apr 12 19:21:53 2014
@@ -35,21 +35,21 @@
     support.</description>
     <inceptionYear>2010</inceptionYear>
     <organization>
-    	<name>The Apache Software Foundation</name>
-    	<url>http://www.apache.org/</url>
+        <name>The Apache Software Foundation</name>
+        <url>http://www.apache.org/</url>
     </organization>
     <scm>
-    	<url>http://svn.apache.org/viewvc/gora/trunk/gora-cassandra/</url>
-    	<connection>scm:svn:http://svn.apache.org/repos/asf/gora/trunk/gora-cassandra/</connection>
-    	<developerConnection>scm:svn:https://svn.apache.org/repos/asf/gora/trunk/gora-cassandra/</developerConnection>
+        <url>http://svn.apache.org/viewvc/gora/trunk/gora-cassandra/</url>
+        <connection>scm:svn:http://svn.apache.org/repos/asf/gora/trunk/gora-cassandra/</connection>
+        <developerConnection>scm:svn:https://svn.apache.org/repos/asf/gora/trunk/gora-cassandra/</developerConnection>
     </scm>
     <issueManagement>
-    	<system>JIRA</system>
-    	<url>https://issues.apache.org/jira/browse/GORA</url>
+        <system>JIRA</system>
+        <url>https://issues.apache.org/jira/browse/GORA</url>
     </issueManagement>
     <ciManagement>
-    	<system>Jenkins</system>
-    	<url>https://builds.apache.org/job/Gora-trunk/</url>
+        <system>Jenkins</system>
+        <url>https://builds.apache.org/job/Gora-trunk/</url>
     </ciManagement>
 
     <properties>
@@ -127,15 +127,15 @@
             <exclusions>
                 <exclusion>
                     <groupId>org.apache.cassandra.deps</groupId>
-    		    <artifactId>avro</artifactId>
+                    <artifactId>avro</artifactId>
                 </exclusion>
             </exclusions>
         </dependency>
         
-        <dependency>
+        <!-- dependency>
             <groupId>org.apache.cassandra</groupId>
             <artifactId>cassandra-thrift</artifactId>
-        </dependency>
+        </dependency-->
 
         <dependency>
             <groupId>org.hectorclient</groupId>
@@ -143,7 +143,7 @@
                 <exclusions>
                     <exclusion>
                         <groupId>org.apache.cassandra</groupId>
-    		        <artifactId>cassandra-all</artifactId>
+                        <artifactId>cassandra-all</artifactId>
                     </exclusion>
             </exclusions>
         </dependency>
@@ -159,11 +159,13 @@
             <artifactId>jdom</artifactId>
         </dependency>
 
-        <!-- Logging Dependencies -->
+        
         <dependency>
-            <groupId>org.apache.hadoop</groupId>
+            <groupId>org.apache.avro</groupId>
             <artifactId>avro</artifactId>
         </dependency>
+        
+        <!-- Logging Dependencies -->
         <dependency>
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-api</artifactId>

Modified: gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraColumn.java
URL: http://svn.apache.org/viewvc/gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraColumn.java?rev=1586888&r1=1586887&r2=1586888&view=diff
==============================================================================
--- gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraColumn.java (original)
+++ gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraColumn.java Sat Apr 12 19:21:53 2014
@@ -18,12 +18,19 @@
 
 package org.apache.gora.cassandra.query;
 
+import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.util.concurrent.ConcurrentHashMap;
 
 import me.prettyprint.hector.api.Serializer;
 
 import org.apache.avro.Schema;
 import org.apache.avro.Schema.Field;
+import org.apache.avro.Schema.Type;
+import org.apache.avro.io.BinaryDecoder;
+import org.apache.avro.io.DecoderFactory;
+import org.apache.avro.specific.SpecificDatumReader;
+import org.apache.gora.cassandra.serializers.AvroSerializerUtil;
 import org.apache.gora.cassandra.serializers.GoraSerializerTypeInferer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -41,7 +48,7 @@ public abstract class CassandraColumn {
   private int type;
   private Field field;
   private int unionType;
-
+  
   public void setUnionType(int pUnionType){
     this.unionType = pUnionType;
   }
@@ -72,7 +79,7 @@ public abstract class CassandraColumn {
 
   public abstract ByteBuffer getName();
   public abstract Object getValue();
-
+  
   protected Object fromByteBuffer(Schema schema, ByteBuffer byteBuffer) {
     Object value = null;
     Serializer<?> serializer = GoraSerializerTypeInferer.getSerializer(schema);
@@ -81,8 +88,14 @@ public abstract class CassandraColumn {
           + "could be found. Please report this to dev@gora.apache.org");
     } else {
       value = serializer.fromByteBuffer(byteBuffer);
+      if (schema.getType().equals(Type.RECORD) || schema.getType().equals(Type.MAP) ){
+        try {
+          value = AvroSerializerUtil.deserializer(value, schema);
+        } catch (IOException e) {
+          LOG.warn(field.name() + " named field could not be deserialized.");
+        }
+      }
     }
     return value;
   }
-
 }

Modified: gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraResult.java
URL: http://svn.apache.org/viewvc/gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraResult.java?rev=1586888&r1=1586887&r2=1586888&view=diff
==============================================================================
--- gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraResult.java (original)
+++ gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraResult.java Sat Apr 12 19:21:53 2014
@@ -72,7 +72,7 @@ public class CassandraResult<K, T extend
     
     for (int iCnt = 0; iCnt < pCassandraRow.length; iCnt++){
       CassandraColumn cColumn = (CassandraColumn)pCassandraRow[iCnt];
-      String columnName = StringSerializer.get().fromByteBuffer(cColumn.getName());
+      String columnName = StringSerializer.get().fromByteBuffer(cColumn.getName().duplicate());
       if (pFieldName.equals(columnName))
         return cColumn;
     }
@@ -95,37 +95,50 @@ public class CassandraResult<K, T extend
     List<Field> fields = schema.getFields();
     
     for (CassandraColumn cassandraColumn: cassandraRow) {
-      
       // get field name
-      String family = cassandraColumn.getFamily();
-      String fieldName = this.reverseMap.get(family + ":" + StringSerializer.get().fromByteBuffer(cassandraColumn.getName()));
+      String family = cassandraColumn.getFamily();  
+      
+      String fieldName = this.reverseMap.get(family + ":" + StringSerializer.get().fromByteBuffer(cassandraColumn.getName().duplicate()));
       
-      if (fieldName != null ){
+      if (fieldName != null) {
         // get field
-        int pos = this.persistent.getFieldIndex(fieldName);
-        Field field = fields.get(pos);
-        Type fieldType = field.schema().getType();
-        System.out.println(StringSerializer.get().fromByteBuffer(cassandraColumn.getName()) + fieldName + " " + fieldType.name());
-        if (fieldType == Type.UNION){
-          // TODO getting UNION stored type
-          // TODO get value of UNION stored type. This field does not need to be written back to the store
-          cassandraColumn.setUnionType(getNonNullTypePos(field.schema().getTypes()));
-        }
+        if (fieldName.indexOf(CassandraStore.UNION_COL_SUFIX) < 0) {
 
-        // get value
-        cassandraColumn.setField(field);
-        Object value = cassandraColumn.getValue();
-
-        this.persistent.put(pos, value);
-        // this field does not need to be written back to the store
-        this.persistent.clearDirty(pos);
-      }
-      else
+          int pos = this.persistent.getSchema().getField(fieldName).pos();
+          Field field = fields.get(pos);
+          Type fieldType = field.schema().getType();
+          // LOG.info(StringSerializer.get().fromByteBuffer(cassandraColumn.getName())
+          // + fieldName + " " + fieldType.name());
+          if (fieldType.equals(Type.UNION)) {
+            //getting UNION stored type
+            CassandraColumn cc = getUnionTypeColumn(fieldName
+                + CassandraStore.UNION_COL_SUFIX, cassandraRow.toArray());
+            //creating temporary UNION Field
+            Field unionField = new Field(fieldName
+                + CassandraStore.UNION_COL_SUFIX, Schema.create(Type.INT),
+                null, null);
+            // get value of UNION stored type
+            cc.setField(unionField);
+            Object val = cc.getValue();
+            cassandraColumn.setUnionType(Integer.parseInt(val.toString()));
+          }
+
+          // get value
+          cassandraColumn.setField(field);
+          Object value = cassandraColumn.getValue();
+
+          this.persistent.put(pos, value);
+          // this field does not need to be written back to the store
+          this.persistent.clearDirty(pos);
+        }
+      } else
         LOG.debug("FieldName was null while iterating CassandraRow and using Avro Union type");
     }
 
   }
 
+  //TODO Should we remove this method?
+  @SuppressWarnings("unused")
   private int getNonNullTypePos(List<Schema> pTypes){
     int iCnt = 0;
     for (Schema sch :  pTypes)

Modified: gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraSubColumn.java
URL: http://svn.apache.org/viewvc/gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraSubColumn.java?rev=1586888&r1=1586887&r2=1586888&view=diff
==============================================================================
--- gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraSubColumn.java (original)
+++ gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraSubColumn.java Sat Apr 12 19:21:53 2014
@@ -19,40 +19,23 @@
 package org.apache.gora.cassandra.query;
 
 import java.nio.ByteBuffer;
-import java.nio.CharBuffer;
-import java.nio.charset.CharacterCodingException;
-import java.nio.charset.Charset;
-import java.nio.charset.CharsetEncoder;
-
-import me.prettyprint.cassandra.serializers.FloatSerializer;
-import me.prettyprint.cassandra.serializers.DoubleSerializer;
-import me.prettyprint.cassandra.serializers.IntegerSerializer;
-import me.prettyprint.cassandra.serializers.LongSerializer;
-import me.prettyprint.cassandra.serializers.StringSerializer;
+import java.util.List;
+import java.util.Map;
+
 import me.prettyprint.hector.api.beans.HColumn;
 
 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;
-import org.apache.avro.util.Utf8;
-import org.apache.gora.cassandra.serializers.GenericArraySerializer;
-import org.apache.gora.cassandra.serializers.StatefulHashMapSerializer;
-import org.apache.gora.cassandra.serializers.TypeUtils;
+import org.apache.gora.cassandra.serializers.ListSerializer;
+import org.apache.gora.cassandra.serializers.MapSerializer;
 import org.apache.gora.cassandra.store.CassandraStore;
-import org.apache.gora.persistency.StatefulHashMap;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 public class CassandraSubColumn extends CassandraColumn {
   public static final Logger LOG = LoggerFactory.getLogger(CassandraSubColumn.class);
 
-  private static final String ENCODING = "UTF-8";
-  
-  private static CharsetEncoder charsetEncoder = Charset.forName(ENCODING).newEncoder();;
-
-
   /**
    * Key-value pair containing the raw data.
    */
@@ -62,6 +45,32 @@ public class CassandraSubColumn extends 
     return hColumn.getName();
   }
 
+  private Object getFieldValue(Type type, Schema fieldSchema, ByteBuffer byteBuffer){
+    Object value = null;
+    if (type.equals(Type.ARRAY)) {
+      ListSerializer<?> serializer = ListSerializer.get(fieldSchema.getElementType());
+      List<?> genericArray = serializer.fromByteBuffer(byteBuffer);
+      value = genericArray;
+    } else if (type.equals(Type.MAP)) {
+//      MapSerializer<?> serializer = MapSerializer.get(fieldSchema.getValueType());
+//      Map<?, ?> map = serializer.fromByteBuffer(byteBuffer);
+//      value = map;
+      value = fromByteBuffer(fieldSchema, byteBuffer);
+    } else if (type.equals(Type.RECORD)){
+      value = fromByteBuffer(fieldSchema, byteBuffer);
+    } else if (type.equals(Type.UNION)){
+      // the selected union schema is obtained
+      Schema unionFieldSchema = getUnionSchema(super.getUnionType(), fieldSchema);
+      Type unionFieldType = unionFieldSchema.getType();
+      // we use the selected union schema to deserialize our actual value
+      //value = fromByteBuffer(unionFieldSchema, byteBuffer);
+      value = getFieldValue(unionFieldType, unionFieldSchema, byteBuffer);
+    } else {
+      value = fromByteBuffer(fieldSchema, byteBuffer);
+    }
+    return value;
+  }
+
   /**
    * Deserialize a String into an typed Object, according to the field schema.
    * @see org.apache.gora.cassandra.query.CassandraColumn#getValue()
@@ -74,24 +83,8 @@ public class CassandraSubColumn extends 
     if (byteBuffer == null) {
       return null;
     }
-    Object value = null;
-    if (type == Type.ARRAY) {
-      GenericArraySerializer serializer = GenericArraySerializer.get(fieldSchema.getElementType());
-      GenericArray genericArray = serializer.fromByteBuffer(byteBuffer);
-      value = genericArray;
-    } else if (type == Type.MAP) {
-      StatefulHashMapSerializer serializer = StatefulHashMapSerializer.get(fieldSchema.getValueType());
-      StatefulHashMap map = serializer.fromByteBuffer(byteBuffer);
-      value = map;
-    } else if (type == Type.UNION){
-      // the selected union schema is obtained
-      Schema unionFieldSchema = getUnionSchema(super.getUnionType(), field.schema());
-      // we use the selected union schema to deserialize our actual value
-      value = fromByteBuffer(unionFieldSchema, byteBuffer);
-    } else {
-      value = fromByteBuffer(fieldSchema, byteBuffer);
-    }
 
+    Object value = getFieldValue(type, fieldSchema, byteBuffer);
     return value;
   }
   

Modified: gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraSuperColumn.java
URL: http://svn.apache.org/viewvc/gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraSuperColumn.java?rev=1586888&r1=1586887&r2=1586888&view=diff
==============================================================================
--- gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraSuperColumn.java (original)
+++ gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraSuperColumn.java Sat Apr 12 19:21:53 2014
@@ -19,8 +19,12 @@
 package org.apache.gora.cassandra.query;
 
 import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 
+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;
@@ -29,9 +33,8 @@ import org.apache.avro.Schema;
 import org.apache.avro.Schema.Field;
 import org.apache.avro.Schema.Type;
 import org.apache.avro.util.Utf8;
-import org.apache.gora.cassandra.serializers.Utf8Serializer;
-import org.apache.gora.persistency.ListGenericArray;
-import org.apache.gora.persistency.StatefulHashMap;
+import org.apache.gora.cassandra.serializers.CharSequenceSerializer;
+import org.apache.gora.cassandra.store.CassandraStore;
 import org.apache.gora.persistency.impl.PersistentBase;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -45,19 +48,14 @@ public class CassandraSuperColumn extend
     return StringSerializer.get().toByteBuffer(hSuperColumn.getName());
   }
 
-  public Object getValue() {
-    Field field = getField();
-    Schema fieldSchema = field.schema();
-    Type type = fieldSchema.getType();
-    
+ private Object getSuperValue(Field field, Schema fieldSchema, Type type){
     Object value = null;
     
     switch (type) {
       case ARRAY:
-        ListGenericArray array = new ListGenericArray(fieldSchema.getElementType());
+        List<Object> array = new ArrayList<Object>();
         
         for (HColumn<ByteBuffer, ByteBuffer> hColumn : this.hSuperColumn.getColumns()) {
-          ByteBuffer memberByteBuffer = hColumn.getValue();
           Object memberValue = fromByteBuffer(fieldSchema.getElementType(), hColumn.getValue());
           // int i = IntegerSerializer().get().fromByteBuffer(hColumn.getName());
           array.add(memberValue);      
@@ -66,13 +64,26 @@ public class CassandraSuperColumn extend
         
         break;
       case MAP:
-        Map<Utf8, Object> map = new StatefulHashMap<Utf8, Object>();
-        
+        Map<CharSequence, Object> map = new HashMap<CharSequence, Object>();
+
         for (HColumn<ByteBuffer, ByteBuffer> hColumn : this.hSuperColumn.getColumns()) {
-          ByteBuffer memberByteBuffer = hColumn.getValue();
-          Object memberValue = null;
-          memberValue = fromByteBuffer(fieldSchema.getValueType(), hColumn.getValue());
-          map.put(Utf8Serializer.get().fromByteBuffer(hColumn.getName()), memberValue);      
+          CharSequence mapKey = CharSequenceSerializer.get().fromByteBuffer(hColumn.getName());
+          if (mapKey.toString().indexOf(CassandraStore.UNION_COL_SUFIX) < 0) {
+            Object memberValue = null;
+            // We need detect real type for UNION Fields
+            if (fieldSchema.getValueType().getType().equals(Type.UNION)){
+              
+              HColumn<ByteBuffer, ByteBuffer> cc = getUnionTypeColumn(mapKey
+                  + CassandraStore.UNION_COL_SUFIX, this.hSuperColumn.getColumns());
+              Integer unionIndex = getUnionIndex(mapKey.toString(), cc);
+              Schema realSchema = fieldSchema.getValueType().getTypes().get(unionIndex);
+              memberValue = fromByteBuffer(realSchema, hColumn.getValue());
+              
+            }else{
+              memberValue = fromByteBuffer(fieldSchema.getValueType(), hColumn.getValue());            
+            }            
+            map.put(mapKey, memberValue);      
+          }
         }
         value = map;
         
@@ -104,21 +115,77 @@ public class CassandraSuperColumn extend
 
           for (HColumn<ByteBuffer, ByteBuffer> hColumn : this.hSuperColumn.getColumns()) {
             String memberName = StringSerializer.get().fromByteBuffer(hColumn.getName());
+            if (memberName.indexOf(CassandraStore.UNION_COL_SUFIX) < 0) {
+              
             if (memberName == null || memberName.length() == 0) {
               LOG.warn("member name is null or empty.");
               continue;
             }
             Field memberField = fieldSchema.getField(memberName);
+            Schema memberSchema = memberField.schema();
+            Type memberType = memberSchema.getType();
+            
             CassandraSubColumn cassandraColumn = new CassandraSubColumn();
             cassandraColumn.setField(memberField);
             cassandraColumn.setValue(hColumn);
-            record.put(record.getFieldIndex(memberName), cassandraColumn.getValue());
+            
+            if (memberType.equals(Type.UNION)){
+              HColumn<ByteBuffer, ByteBuffer> hc = getUnionTypeColumn(memberField.name()
+                  + CassandraStore.UNION_COL_SUFIX, this.hSuperColumn.getColumns().toArray());
+              Integer unionIndex = getUnionIndex(memberField.name(),hc);
+              cassandraColumn.setUnionType(unionIndex);
+            }
+            
+            record.put(record.getSchema().getField(memberName).pos(), cassandraColumn.getValue());
+          }
           }
         }
         break;
+      case UNION:
+        int schemaPos = this.getUnionType();
+        Schema unioSchema = fieldSchema.getTypes().get(schemaPos);
+        Type unionType = unioSchema.getType();
+        value = getSuperValue(field, unioSchema, unionType);
+        break;
       default:
-        LOG.warn("Type: " + type.name() + " not supported for field: " + field.name() + ". Please report this to dev@gora.apache.org");
+        Object memberValue = null;
+        // Using for UnionIndex of Union type field get value. UnionIndex always Integer.  
+        for (HColumn<ByteBuffer, ByteBuffer> hColumn : this.hSuperColumn.getColumns()) {
+          memberValue = fromByteBuffer(fieldSchema, hColumn.getValue());      
+        }
+        value = memberValue;
+        LOG.warn("Type: " + type.name() + " not supported for field: " + field.name());
     }
+    return value;
+  }
+
+ private Integer getUnionIndex(String fieldName, HColumn<ByteBuffer, ByteBuffer> uc){
+   Integer val = IntegerSerializer.get().fromByteBuffer(uc.getValue());
+   return Integer.parseInt(val.toString());
+ }
+ 
+  private HColumn<ByteBuffer, ByteBuffer> getUnionTypeColumn(String fieldName,
+    List<HColumn<ByteBuffer, ByteBuffer>> columns) {
+    return getUnionTypeColumn(fieldName, columns.toArray());
+}
+
+  private HColumn<ByteBuffer, ByteBuffer> getUnionTypeColumn(String fieldName, Object[] hColumns) {
+    for (int iCnt = 0; iCnt < hColumns.length; iCnt++){
+      @SuppressWarnings("unchecked")
+      HColumn<ByteBuffer, ByteBuffer> hColumn = (HColumn<ByteBuffer, ByteBuffer>)hColumns[iCnt];
+      String columnName = StringSerializer.get().fromByteBuffer(hColumn.getNameBytes().duplicate());
+      if (fieldName.equals(columnName))
+        return hColumn;
+    }
+    return null;
+}
+
+  public Object getValue() {
+    Field field = getField();
+    Schema fieldSchema = field.schema();
+    Type type = fieldSchema.getType();
+    
+    Object value = getSuperValue(field, fieldSchema, type);
     
     return value;
   }

Added: gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/AvroSerializerUtil.java
URL: http://svn.apache.org/viewvc/gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/AvroSerializerUtil.java?rev=1586888&view=auto
==============================================================================
--- gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/AvroSerializerUtil.java (added)
+++ gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/AvroSerializerUtil.java Sat Apr 12 19:21:53 2014
@@ -0,0 +1,94 @@
+package org.apache.gora.cassandra.serializers;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.avro.Schema;
+import org.apache.avro.io.BinaryDecoder;
+import org.apache.avro.io.BinaryEncoder;
+import org.apache.avro.io.DecoderFactory;
+import org.apache.avro.io.EncoderFactory;
+import org.apache.avro.specific.SpecificDatumReader;
+import org.apache.avro.specific.SpecificDatumWriter;
+
+public class AvroSerializerUtil {
+
+  /**
+   * Threadlocals maintaining reusable binary decoders and encoders.
+   */
+  private static ThreadLocal<ByteArrayOutputStream> outputStream =
+      new ThreadLocal<ByteArrayOutputStream>();
+  
+  public static final ThreadLocal<BinaryEncoder> encoders =
+      new ThreadLocal<BinaryEncoder>();
+
+  public static final ThreadLocal<BinaryDecoder> decoders =
+      new ThreadLocal<BinaryDecoder>();
+  
+  /**
+   * 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<String, SpecificDatumWriter<?>>();  
+  
+  public static final ConcurrentHashMap<String, SpecificDatumReader<?>> readerMap = 
+      new ConcurrentHashMap<String, SpecificDatumReader<?>>();
+  
+  @SuppressWarnings({ "unchecked", "rawtypes" })
+  public static <T> byte[] serializer(T value, Schema schema) throws IOException{
+    SpecificDatumWriter writer = (SpecificDatumWriter<?>) writerMap.get(schema.getFullName());
+    if (writer == null) {
+      writer = new SpecificDatumWriter(schema);// ignore dirty bits
+      writerMap.put(schema.getFullName(),writer);
+    }
+    
+    BinaryEncoder encoderFromCache = encoders.get();
+    ByteArrayOutputStream bos = new ByteArrayOutputStream();
+    outputStream.set(bos);
+    BinaryEncoder encoder = EncoderFactory.get().directBinaryEncoder(bos, null);
+    if (encoderFromCache == null) {
+      encoders.set(encoder);
+    }
+    
+    //reset the buffers
+    ByteArrayOutputStream os = outputStream.get();
+    os.reset();
+    
+    writer.write(value, encoder);
+    encoder.flush();
+    byte[] byteValue = os.toByteArray();
+    return byteValue;
+  }
+  
+  public static Object deserializer(Object value, Schema schema) throws IOException{
+    String schemaId = schema.getFullName();      
+    
+    SpecificDatumReader<?> reader = (SpecificDatumReader<?>)readerMap.get(schemaId);
+    if (reader == null) {
+      reader = new SpecificDatumReader(schema);// ignore dirty bits
+      SpecificDatumReader localReader=null;
+      if((localReader=readerMap.putIfAbsent(schemaId, reader))!=null) {
+        reader = localReader;
+      }
+    }
+    
+    // initialize a decoder, possibly reusing previous one
+    BinaryDecoder decoderFromCache = decoders.get();
+    BinaryDecoder decoder = DecoderFactory.get().binaryDecoder((byte[])value, null);
+    // put in threadlocal cache if the initial get was empty
+    if (decoderFromCache==null) {
+      decoders.set(decoder);
+    }
+
+    Object result = reader.read(null, decoder);
+    return result;
+
+  }
+}

Propchange: gora/trunk/gora-cassandra/src/main/java/org/apache/gora/cassandra/serializers/AvroSerializerUtil.java
------------------------------------------------------------------------------
    svn:mime-type = text/plain



Mime
View raw message