Return-Path: X-Original-To: archive-asf-public-internal@cust-asf2.ponee.io Delivered-To: archive-asf-public-internal@cust-asf2.ponee.io Received: from cust-asf.ponee.io (cust-asf.ponee.io [163.172.22.183]) by cust-asf2.ponee.io (Postfix) with ESMTP id 13648200C87 for ; Wed, 17 May 2017 22:44:55 +0200 (CEST) Received: by cust-asf.ponee.io (Postfix) id 11EB0160BCB; Wed, 17 May 2017 20:44:55 +0000 (UTC) Delivered-To: archive-asf-public@cust-asf.ponee.io Received: from mail.apache.org (hermes.apache.org [140.211.11.3]) by cust-asf.ponee.io (Postfix) with SMTP id 58C9D160BAC for ; Wed, 17 May 2017 22:44:54 +0200 (CEST) Received: (qmail 15788 invoked by uid 500); 17 May 2017 20:44:52 -0000 Mailing-List: contact commits-help@pig.apache.org; run by ezmlm Precedence: bulk List-Help: List-Unsubscribe: List-Post: List-Id: Reply-To: dev@pig.apache.org Delivered-To: mailing list commits@pig.apache.org Received: (qmail 15778 invoked by uid 99); 17 May 2017 20:44:51 -0000 Received: from Unknown (HELO svn01-us-west.apache.org) (209.188.14.144) by apache.org (qpsmtpd/0.29) with ESMTP; Wed, 17 May 2017 20:44:51 +0000 Received: from svn01-us-west.apache.org (localhost [127.0.0.1]) by svn01-us-west.apache.org (ASF Mail Server at svn01-us-west.apache.org) with ESMTP id 1097D3A0460 for ; Wed, 17 May 2017 20:44:50 +0000 (UTC) Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit Subject: svn commit: r1795450 - in /pig/branches/spark: src/org/apache/pig/impl/util/avro/AvroTupleWrapper.java test/excluded-tests-spark Date: Wed, 17 May 2017 20:44:50 -0000 To: commits@pig.apache.org From: rohini@apache.org X-Mailer: svnmailer-1.0.9 Message-Id: <20170517204451.1097D3A0460@svn01-us-west.apache.org> archived-at: Wed, 17 May 2017 20:44:55 -0000 Author: rohini Date: Wed May 17 20:44:50 2017 New Revision: 1795450 URL: http://svn.apache.org/viewvc?rev=1795450&view=rev Log: PIG-5134: Fix TestAvroStorage unit test in Spark mode (nkollar via rohini) Modified: pig/branches/spark/src/org/apache/pig/impl/util/avro/AvroTupleWrapper.java pig/branches/spark/test/excluded-tests-spark Modified: pig/branches/spark/src/org/apache/pig/impl/util/avro/AvroTupleWrapper.java URL: http://svn.apache.org/viewvc/pig/branches/spark/src/org/apache/pig/impl/util/avro/AvroTupleWrapper.java?rev=1795450&r1=1795449&r2=1795450&view=diff ============================================================================== --- pig/branches/spark/src/org/apache/pig/impl/util/avro/AvroTupleWrapper.java (original) +++ pig/branches/spark/src/org/apache/pig/impl/util/avro/AvroTupleWrapper.java Wed May 17 20:44:50 2017 @@ -26,8 +26,17 @@ 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.generic.GenericDatumReader; +import org.apache.avro.generic.GenericDatumWriter; import org.apache.avro.generic.GenericEnumSymbol; import org.apache.avro.generic.IndexedRecord; +import org.apache.avro.io.DatumReader; +import org.apache.avro.io.DatumWriter; +import org.apache.avro.io.Decoder; +import org.apache.avro.io.DecoderFactory; +import org.apache.avro.io.Encoder; +import org.apache.avro.io.EncoderFactory; +import org.apache.avro.util.Utf8; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.pig.backend.executionengine.ExecException; @@ -38,6 +47,8 @@ import org.apache.pig.data.TupleFactory; import java.io.DataInput; import java.io.DataOutput; import java.io.IOException; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; import java.nio.ByteBuffer; import java.util.Iterator; import java.util.List; @@ -50,12 +61,12 @@ import java.util.Map; public final class AvroTupleWrapper implements Tuple { private static final Log LOG = LogFactory.getLog(AvroTupleWrapper.class); - private TupleFactory mTupleFactory = TupleFactory.getInstance(); + private transient TupleFactory mTupleFactory = TupleFactory.getInstance(); /** * The Avro object wrapped in the pig Tuple. */ - private T avroObject; + private transient T avroObject; /** * Creates a new AvroTupleWrapper object. @@ -205,7 +216,14 @@ public final class AvroTupleWrapper writer = new GenericDatumWriter<>(); + writer.setSchema(avroObject.getSchema()); + Encoder encoder = EncoderFactory.get().binaryEncoder(out, null); + writer.write(avroObject, encoder); + encoder.flush(); + } + + // Required for Java serialization used by Spark: PIG-5134 + private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException { + Schema schema = new Schema.Parser().parse((String) in.readObject()); + DatumReader reader = new GenericDatumReader<>(schema); + Decoder decoder = DecoderFactory.get().binaryDecoder(in, null); + avroObject = reader.read(avroObject, decoder); + } } Modified: pig/branches/spark/test/excluded-tests-spark URL: http://svn.apache.org/viewvc/pig/branches/spark/test/excluded-tests-spark?rev=1795450&r1=1795449&r2=1795450&view=diff ============================================================================== --- pig/branches/spark/test/excluded-tests-spark (original) +++ pig/branches/spark/test/excluded-tests-spark Wed May 17 20:44:50 2017 @@ -3,7 +3,5 @@ **/TestNativeMapReduce.java **/TestCounters.java -#TODO: PIG-5134 fix for Spark mode -**/TestAvroStorage.java #TODO: PIG-5135 fix for Spark mode **/TestOrcStoragePushdown.java