diff --git a/CHANGES.txt b/CHANGES.txt
index dcef62e..5fb6142 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -187,6 +187,8 @@
 
 BUG FIXES
 
+PIG-5108: AvroStorage on Tez with exception on nested records (daijy)
+
 PIG-4260: SpillableMemoryManager.spill should revert spill on all exception (rohini)
 
 PIG-4918: Pig on Tez cannot switch pig.temp.dir to another fs (daijy)
diff --git a/src/org/apache/pig/impl/util/avro/AvroTupleWrapper.java b/src/org/apache/pig/impl/util/avro/AvroTupleWrapper.java
index 648e639..8adbe84 100644
--- a/src/org/apache/pig/impl/util/avro/AvroTupleWrapper.java
+++ b/src/org/apache/pig/impl/util/avro/AvroTupleWrapper.java
@@ -33,6 +33,7 @@
 import org.apache.pig.backend.executionengine.ExecException;
 import org.apache.pig.data.DataByteArray;
 import org.apache.pig.data.Tuple;
+import org.apache.pig.data.TupleFactory;
 
 import java.io.DataInput;
 import java.io.DataOutput;
@@ -49,6 +50,7 @@
 public final class AvroTupleWrapper <T extends IndexedRecord>
     implements Tuple {
     private static final Log LOG = LogFactory.getLog(AvroTupleWrapper.class);
+    private TupleFactory mTupleFactory = TupleFactory.getInstance();
 
   /**
    * The Avro object wrapped in the pig Tuple.
@@ -64,9 +66,9 @@
   }
 
   @Override
-  public void write(final DataOutput o) throws IOException {
-    throw new IOException(
-        this.getClass().toString() + ".write called, but not implemented yet");
+  public void write(DataOutput out) throws IOException {
+      Tuple t = mTupleFactory.newTupleNoCopy(getAll());
+      t.write(out);
   }
 
   @SuppressWarnings("rawtypes")
diff --git a/test/org/apache/pig/builtin/TestAvroStorage.java b/test/org/apache/pig/builtin/TestAvroStorage.java
index ee12ccb..531bc26 100644
--- a/test/org/apache/pig/builtin/TestAvroStorage.java
+++ b/test/org/apache/pig/builtin/TestAvroStorage.java
@@ -709,6 +709,19 @@
     }
 
     @Test
+    public void testGroupWithRepeatedSubRecords() throws Exception {
+      final String input = basedir + "data/avro/uncompressed/recordWithRepeatedSubRecords.avro";
+      final String check = basedir + "data/avro/uncompressed/recordWithRepeatedSubRecords.avro";
+      testAvroStorage(true, basedir + "code/pig/group_test.pig",
+          ImmutableMap.of(
+              "INFILE",           input,
+              "AVROSTORAGE_OUT_2", "-f " + basedir + "schema/recordWithRepeatedSubRecords.avsc",
+              "OUTFILE",          createOutputName())
+        );
+      verifyResults(createOutputName(),check);
+    }
+
+    @Test
     public void testLoadDirectory() throws Exception {
       final String input = basedir + "data/avro/uncompressed/testdirectory";
       final String check = basedir + "data/avro/uncompressed/testDirectoryCounts.avro";
diff --git a/test/org/apache/pig/builtin/avro/code/pig/group_test.pig b/test/org/apache/pig/builtin/avro/code/pig/group_test.pig
new file mode 100644
index 0000000..64a9577
--- /dev/null
+++ b/test/org/apache/pig/builtin/avro/code/pig/group_test.pig
@@ -0,0 +1,5 @@
+in = LOAD '$INFILE' USING AvroStorage();
+grouped = GROUP in BY (value1.thing);
+flattened = FOREACH grouped GENERATE flatten(in) as (key: chararray,value1: (thing: chararray,count: int),value2: (thing: chararray,count: int));
+RMF $OUTFILE;
+STORE flattened INTO '$OUTFILE' USING AvroStorage();