You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by da...@apache.org on 2012/06/16 00:00:30 UTC
svn commit: r1350819 - in /pig/trunk: CHANGES.txt
src/org/apache/pig/builtin/BagToString.java
src/org/apache/pig/builtin/BagToTuple.java test/e2e/pig/tests/nightly.conf
test/org/apache/pig/test/TestBuiltInBagToTupleOrString.java
Author: daijy
Date: Fri Jun 15 22:00:29 2012
New Revision: 1350819
URL: http://svn.apache.org/viewvc?rev=1350819&view=rev
Log:
PIG-2166: UDFs to join a bag
Added:
pig/trunk/src/org/apache/pig/builtin/BagToString.java
pig/trunk/src/org/apache/pig/builtin/BagToTuple.java
pig/trunk/test/org/apache/pig/test/TestBuiltInBagToTupleOrString.java
Modified:
pig/trunk/CHANGES.txt
pig/trunk/test/e2e/pig/tests/nightly.conf
Modified: pig/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/pig/trunk/CHANGES.txt?rev=1350819&r1=1350818&r2=1350819&view=diff
==============================================================================
--- pig/trunk/CHANGES.txt (original)
+++ pig/trunk/CHANGES.txt Fri Jun 15 22:00:29 2012
@@ -24,6 +24,8 @@ INCOMPATIBLE CHANGES
IMPROVEMENTS
+PIG-2166: UDFs to join a bag (hluu via daijy)
+
PIG-2651: Provide a much easier to use accumulator interface (jcoveney via daijy)
PIG-2658: Add pig.script.submitted.timestamp and pig.job.submitted.timestamp in generated Map-Reduce job conf (billgraham)
Added: pig/trunk/src/org/apache/pig/builtin/BagToString.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/builtin/BagToString.java?rev=1350819&view=auto
==============================================================================
--- pig/trunk/src/org/apache/pig/builtin/BagToString.java (added)
+++ pig/trunk/src/org/apache/pig/builtin/BagToString.java Fri Jun 15 22:00:29 2012
@@ -0,0 +1,141 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.pig.builtin;
+
+import java.io.IOException;
+
+import org.apache.pig.EvalFunc;
+import org.apache.pig.PigException;
+import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.data.DataBag;
+import org.apache.pig.data.DataType;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.impl.logicalLayer.FrontendException;
+import org.apache.pig.impl.logicalLayer.schema.Schema;
+import org.apache.pig.impl.logicalLayer.schema.Schema.FieldSchema;
+
+/**
+ * Flatten a bag into a string. This UDF will the character '-' as the default delimiter
+ * if one is not provided.
+ *
+ * Example:
+ *
+ * bag = {(a),(b),(c)}
+ * BagToString(bag) -> 'a_b_c'
+ *
+ * BagToString(bag, '+') -> 'a+b+c'
+ *
+ * bag = {(a,b), (c,d), (e,f)}
+ * BagToString(bag) --> 'a_b_c_d_e_f'
+ *
+ * If input bag is null, this UTF will return null;
+ */
+public class BagToString extends EvalFunc<String> {
+
+ private static final String USAGE_STRING = "Usage BagToString(dataBag) or BagToString(dataBag, delimiter)";
+ private static final String DEFAULT_DELIMITER = "_";
+
+ @Override
+ public String exec(Tuple inputTuple) throws IOException {
+ if ((inputTuple.size() != 1) && (inputTuple.size() != 2)) {
+ throw new ExecException(USAGE_STRING, PigException.INPUT);
+ }
+
+ Object firstArg = inputTuple.get(0);
+ if (firstArg == null) {
+ return null;
+ }
+
+ if (!(firstArg instanceof DataBag)) {
+ throw new ExecException(USAGE_STRING + " found type " + firstArg.getClass().getName(), PigException.INPUT);
+ }
+
+ if ((inputTuple.size() == 2) && !(inputTuple.get(1) instanceof String)) {
+ throw new ExecException("Usage BagToTuple(DataBag, String)", PigException.INPUT);
+ }
+
+ DataBag bag = (DataBag) inputTuple.get(0);
+
+ String delimeter = DEFAULT_DELIMITER;
+ if (inputTuple.size() == 2) {
+ delimeter = (String)inputTuple.get(1);
+ }
+ StringBuilder buffer = new StringBuilder();
+
+ try {
+ for (Tuple t : bag) {
+ if (t != null) {
+ for (int i = 0; i < t.size(); i++) {
+ if (buffer.length() > 0) {
+ buffer.append(delimeter);
+ }
+ buffer.append(t.get(i));
+ }
+ }
+ }
+
+ return buffer.toString();
+ } catch (Exception e) {
+ String msg = "Encourntered error while flattening a bag "
+ + this.getClass().getSimpleName();
+ throw new ExecException(msg, PigException.BUG, e);
+ }
+ }
+
+
+ @Override
+ public Schema outputSchema(Schema inputSchema) {
+ try {
+ if ((inputSchema == null) || ((inputSchema.size() != 1) && (inputSchema.size() != 2))) {
+ throw new RuntimeException("Expecting 2 inputs, found: " +
+ ((inputSchema == null) ? 0 : inputSchema.size()));
+ }
+
+ FieldSchema inputFieldSchema = inputSchema.getField(0);
+ if (inputFieldSchema.type != DataType.BAG) {
+ throw new RuntimeException("Expecting a bag of tuples: {()}, found data type: " +
+ DataType.findTypeName(inputFieldSchema.type));
+ }
+
+ // first field in the bag schema
+ FieldSchema firstFieldSchema = inputFieldSchema.schema.getField(0);
+ if ((firstFieldSchema == null) || (firstFieldSchema.schema == null)
+ || firstFieldSchema.schema.size() < 1) {
+ throw new RuntimeException("Expecting a bag and a delimeter, found: " + inputSchema);
+ }
+
+ if (firstFieldSchema.type != DataType.TUPLE) {
+ throw new RuntimeException("Expecting a bag and a delimeter, found: " + inputSchema);
+ }
+
+ if (inputSchema.size() == 2) {
+ FieldSchema secondInputFieldSchema = inputSchema.getField(1);
+
+ if (secondInputFieldSchema.type != DataType.CHARARRAY) {
+ throw new RuntimeException("Expecting a bag and a delimeter, found: " + inputSchema);
+ }
+ }
+
+ return new Schema(new Schema.FieldSchema(null, DataType.CHARARRAY));
+ } catch (FrontendException e) {
+ e.printStackTrace();
+ return null;
+ }
+ }
+
+}
Added: pig/trunk/src/org/apache/pig/builtin/BagToTuple.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/builtin/BagToTuple.java?rev=1350819&view=auto
==============================================================================
--- pig/trunk/src/org/apache/pig/builtin/BagToTuple.java (added)
+++ pig/trunk/src/org/apache/pig/builtin/BagToTuple.java Fri Jun 15 22:00:29 2012
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.pig.builtin;
+
+import java.io.IOException;
+
+import org.apache.pig.EvalFunc;
+import org.apache.pig.PigException;
+import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.data.DataBag;
+import org.apache.pig.data.DataType;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.data.TupleFactory;
+import org.apache.pig.impl.logicalLayer.FrontendException;
+import org.apache.pig.impl.logicalLayer.schema.Schema;
+
+/**
+ * Flatten a bag into a tuple. This UDF performs only flattening at the first level,
+ * it doesn't recursively flatten nested bags.
+ *
+ * Example: {(a),(b),(c)} --> (a,b,c)
+ * {(a,b), (c,d), (e,f)} --> (a,b,c,d,e,f);
+ *
+ * If input bag is null, this UDF will return null;
+ *
+ */
+public class BagToTuple extends EvalFunc<Tuple> {
+
+ @Override
+ public Tuple exec(Tuple inputTuple) throws IOException {
+
+ if (inputTuple.size() != 1) {
+ throw new ExecException("Expecting 1 input, found " + inputTuple.size(), PigException.INPUT);
+ }
+
+ if (inputTuple.get(0) == null) {
+ return null;
+ }
+
+ if (!(inputTuple.get(0) instanceof DataBag)) {
+ throw new ExecException("Usage BagToTuple(DataBag)", PigException.INPUT);
+ }
+
+
+ DataBag inputBag = (DataBag) (inputTuple.get(0));
+ try {
+ Tuple outputTuple = null;
+
+ long outputTupleSize = getOuputTupleSize(inputBag);
+
+ // TupleFactory.newTuple(int size) can only support up to Integer.MAX_VALUE
+ if (outputTupleSize > Integer.MAX_VALUE) {
+ throw new ExecException("Input bag is too large", 105, PigException.INPUT);
+ }
+
+ TupleFactory tupleFactory = TupleFactory.getInstance();
+ outputTuple = tupleFactory.newTuple((int) outputTupleSize);
+
+ int fieldNum = 0;
+ for (Tuple t : inputBag) {
+ if (t != null) {
+ for (int i = 0; i < t.size(); i++) {
+ outputTuple.set(fieldNum++, t.get(i));
+ }
+ }
+ }
+ return outputTuple;
+ } catch (Exception e) {
+ String msg = "Encourntered error while flattening a bag to tuple"
+ + this.getClass().getSimpleName();
+ throw new ExecException(msg, PigException.BUG, e);
+ }
+ }
+
+ /**
+ * Calculate the size of the output tuple based on the sum
+ * of the size of each tuple in the input bag
+ *
+ * @param bag
+ * @return total # of data elements in a tab
+ */
+ private long getOuputTupleSize(DataBag bag) {
+ long size = 0;
+ if (bag != null) {
+ for (Tuple t : bag) {
+ size = size + t.size();
+ }
+ }
+ return size;
+ }
+
+ @Override
+ public Schema outputSchema(Schema inputSchema) {
+ try {
+ if ((inputSchema == null) || inputSchema.size() != 1) {
+ throw new RuntimeException("Expecting 1 input, found " +
+ ((inputSchema == null) ? 0 : inputSchema.size()));
+ }
+
+ Schema.FieldSchema inputFieldSchema = inputSchema.getField(0);
+ if (inputFieldSchema.type != DataType.BAG) {
+ throw new RuntimeException("Expecting a bag of tuples: {()}");
+ }
+
+ // first field in the bag schema
+ Schema.FieldSchema firstFieldSchema = inputFieldSchema.schema.getField(0);
+ if ((firstFieldSchema == null) || (firstFieldSchema.schema == null)
+ || firstFieldSchema.schema.size() < 1) {
+ throw new RuntimeException("Expecting a bag of tuples: {()}, found: " + inputSchema);
+ }
+
+ if (firstFieldSchema.type != DataType.TUPLE) {
+ throw new RuntimeException("Expecting a bag of tuples: {()}, found: " + inputSchema);
+ }
+
+ // now for output schema
+ Schema tupleOutputSchema = new Schema();
+ for (int i = 0; i < firstFieldSchema.schema.size(); ++i) {
+ tupleOutputSchema.add(firstFieldSchema.schema.getField(i));
+ }
+ return new Schema(new Schema.FieldSchema(getSchemaName(this
+ .getClass().getName().toLowerCase(), inputSchema), tupleOutputSchema,
+ DataType.TUPLE));
+ } catch (FrontendException e) {
+ e.printStackTrace();
+ return null;
+ }
+ }
+
+}
Modified: pig/trunk/test/e2e/pig/tests/nightly.conf
URL: http://svn.apache.org/viewvc/pig/trunk/test/e2e/pig/tests/nightly.conf?rev=1350819&r1=1350818&r2=1350819&view=diff
==============================================================================
--- pig/trunk/test/e2e/pig/tests/nightly.conf (original)
+++ pig/trunk/test/e2e/pig/tests/nightly.conf Fri Jun 15 22:00:29 2012
@@ -4842,6 +4842,106 @@ store a into ':OUTPATH:';\,
]
},
{
+ 'name' => 'BagToTuple',
+ 'tests' => [
+ {
+ # basic test of converting bag to tuples. Use the first and last tuple in the bag b
+ 'num' => 1,
+ 'pig' => q\
+ a = load ':INPATH:/singlefile/studentcomplextab10k' using PigStorage() as (m:[], t:(name:chararray, age:int, gpa:double), b:{t:(name:chararray, age:int, gpa:double)});
+ filterA = filter a by b is not null and COUNT(b) > 1;
+ b = foreach filterA {
+ order_desc = order b by age desc;
+ limit_desc = limit order_desc 1;
+ order_asc = order b by age asc;
+ limit_asc = limit order_asc 1;
+ generate FLATTEN(limit_desc), FLATTEN(limit_asc);
+ };
+ c = foreach b generate TOTUPLE(limit_desc::name, limit_desc::age, limit_desc::gpa) as first_t,
+ TOTUPLE(limit_asc::name, limit_asc::age, limit_asc::gpa) as second_t;
+ d = foreach c generate TOBAG(first_t, second_t) as n_bag;
+ e = foreach d generate BagToTuple(n_bag);
+ store e into ':OUTPATH:';\,
+ 'verify_pig_script' => q\
+ a = load ':INPATH:/singlefile/studentcomplextab10k' using PigStorage() as (m:[], t:(name:chararray, age:int, gpa:double), b:{t:(name:chararray, age:int, gpa:double)});
+ filterA = filter a by b is not null and COUNT(b) > 1;
+ b = foreach filterA {
+ order_desc = order b by age desc;
+ limit_desc = limit order_desc 1;
+ order_asc = order b by age asc;
+ limit_asc = limit order_asc 1;
+ generate FLATTEN(limit_desc), FLATTEN(limit_asc);
+ };
+ c = foreach b generate TOTUPLE(limit_desc::name, limit_desc::age, limit_desc::gpa, limit_asc::name,limit_asc::age, limit_asc::gpa) as big_t;
+ store c into ':OUTPATH:';\,
+ },
+ {
+ # covert an existing tuple to bag and use the output of BagToTuple
+ 'num' => 2,
+ 'pig' => q\
+ a = load ':INPATH:/singlefile/studentcomplextab10k' using PigStorage() as (m:[], t:(name:chararray, age:int, gpa:double), b:{t:(name:chararray, age:int, gpa:double)});
+ b = filter a by t is not null;
+ c = foreach b generate TOBAG(t) as newBag;
+ d = foreach c generate BagToTuple(newBag);
+ store d into ':OUTPATH:';\,
+ 'verify_pig_script' => q\
+ a = load ':INPATH:/singlefile/studentcomplextab10k' using PigStorage() as (m:[], t:(name:chararray, age:int, gpa:double), b:{t:(name:chararray, age:int, gpa:double)});
+ b = filter a by t is not null;
+ c = foreach b generate t;
+ store c into ':OUTPATH:';\,
+ },
+ ]
+ },
+ {
+ 'name' => 'BagToString',
+ 'tests' => [
+ {
+ 'num' => 1,
+ 'pig' => q\
+ a = load ':INPATH:/singlefile/studentcomplextab10k' using PigStorage() as (m:[], t:(name:chararray, age:int, gpa:double), b:{t:(name:chararray, age:int, gpa:double)});
+ filterA = filter a by b is not null and COUNT(b) > 1;
+ b = foreach filterA {
+ order_desc = order b by age desc;
+ limit_desc = limit order_desc 1;
+ order_asc = order b by age asc;
+ limit_asc = limit order_asc 1;
+ generate FLATTEN(limit_desc), FLATTEN(limit_asc);
+ };
+ c = foreach b generate TOTUPLE(limit_desc::name, limit_desc::age, limit_desc::gpa) as first_t,
+ TOTUPLE(limit_asc::name, limit_asc::age, limit_asc::gpa) as second_t;
+ d = foreach c generate TOBAG(first_t, second_t) as n_bag;
+ e = foreach d generate BagToString(n_bag);
+ store e into ':OUTPATH:';\,
+ 'verify_pig_script' => q\
+ a = load ':INPATH:/singlefile/studentcomplextab10k' using PigStorage() as (m:[], t:(name:chararray, age:int, gpa:double), b:{t:(name:chararray, age:int, gpa:double)});
+ filterA = filter a by b is not null and COUNT(b) > 1;
+ b = foreach filterA {
+ order_desc = order b by age desc;
+ limit_desc = limit order_desc 1;
+ order_asc = order b by age asc;
+ limit_asc = limit order_asc 1;
+ generate FLATTEN(limit_desc), FLATTEN(limit_asc);
+ };
+ c = foreach b generate CONCAT(limit_desc::name, CONCAT('_', CONCAT((chararray)limit_desc::age, CONCAT('_', CONCAT((chararray)limit_desc::gpa, CONCAT('_',CONCAT(limit_asc::name,CONCAT('_',CONCAT((chararray)limit_asc::age, CONCAT('_',(chararray)limit_asc::gpa)))))))))) as big_t;
+ store c into ':OUTPATH:';\,
+ },
+ {
+ 'num' => 2,
+ 'pig' => q\
+ a = load ':INPATH:/singlefile/studentcomplextab10k' using PigStorage() as (m:[], t:(name:chararray, age:int, gpa:double), b:{t:(name:chararray, age:int, gpa:double)});
+ b = filter a by t is not null;
+ c = foreach b generate TOBAG(t) as newBag;
+ d = foreach c generate BagToString(newBag);
+ store d into ':OUTPATH:';\,
+ 'verify_pig_script' => q\
+ a = load ':INPATH:/singlefile/studentcomplextab10k' using PigStorage() as (m:[], t:(name:chararray, age:int, gpa:double), b:{t:(name:chararray, age:int, gpa:double)});
+ b = filter a by t is not null;
+ c = foreach b generate CONCAT(t.name, CONCAT('_', CONCAT((chararray)t.age, CONCAT('_', (chararray)t.gpa))));
+ store c into ':OUTPATH:';\,
+ },
+ ]
+ },
+ {
'name' => 'NestedCross',
'tests' => [
{
Added: pig/trunk/test/org/apache/pig/test/TestBuiltInBagToTupleOrString.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestBuiltInBagToTupleOrString.java?rev=1350819&view=auto
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestBuiltInBagToTupleOrString.java (added)
+++ pig/trunk/test/org/apache/pig/test/TestBuiltInBagToTupleOrString.java Fri Jun 15 22:00:29 2012
@@ -0,0 +1,596 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.pig.test;
+
+import static junit.framework.Assert.*;
+
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.pig.ExecType;
+import org.apache.pig.PigServer;
+import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.builtin.BagToString;
+import org.apache.pig.builtin.BagToTuple;
+import org.apache.pig.builtin.mock.Storage.Data;
+import org.apache.pig.data.BagFactory;
+import org.apache.pig.data.DataBag;
+import org.apache.pig.data.DataType;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.data.TupleFactory;
+import org.apache.pig.impl.logicalLayer.schema.Schema;
+import org.apache.pig.impl.logicalLayer.schema.Schema.FieldSchema;
+import static org.apache.pig.builtin.mock.Storage.*;
+
+import org.junit.Test;
+
+/**
+ *
+ * Test cases for BagToTuple and BagToString UDFs
+ *
+ * @author hluu
+ *
+ */
+public class TestBuiltInBagToTupleOrString {
+ private BagFactory bf = BagFactory.getInstance();
+ private TupleFactory tf = TupleFactory.getInstance();
+
+ @Test
+ public void testNullInputBagToTupleUDF() throws Exception {
+ BagToTuple udf = new BagToTuple();
+ Tuple udfInput = tf.newTuple(1);
+ udfInput.set(0, null);
+ Tuple output = udf.exec(udfInput);
+ assertNull(output);
+ }
+
+ @Test
+ public void testBasicBagToTupleUDF() throws Exception {
+
+ Tuple t1 = tf.newTuple(2);
+ t1.set(0, "a");
+ t1.set(1, 5);
+
+ Tuple t2 = tf.newTuple(2);
+ t2.set(0, "c");
+ t2.set(1, 6);
+
+ DataBag bag = bf.newDefaultBag();
+ bag.add(t1);
+ bag.add(t2);
+
+ Tuple udfInput = tf.newTuple(1);
+ udfInput.set(0, bag);
+
+ // invoking UDF
+ BagToTuple udf = new BagToTuple();
+ Tuple result = udf.exec(udfInput);
+
+ int totalExpectedSize = t1.size() + t2.size();
+ assertEquals(totalExpectedSize, result.size());
+
+ for (int i = 0; i < t1.size(); i++) {
+ assertEquals(t1.get(i), result.get(i));
+ }
+
+ for (int i = 0; i < t2.size(); i++) {
+ assertEquals(t2.get(i), result.get(t1.size() + i));
+ }
+ }
+
+ @Test
+ public void testNonuniformTuplesInBagForBagToTupleUDF() throws Exception {
+
+ Tuple t1 = tf.newTuple(2);
+ t1.set(0, "a");
+ t1.set(1, 5);
+
+ Tuple t2 = tf.newTuple(3);
+ t2.set(0, "b");
+ t2.set(1, 6);
+ t2.set(2, 7);
+
+ Tuple t3 = tf.newTuple(4);
+ t3.set(0, "c");
+ t3.set(1, 8);
+ t3.set(2, 9.7);
+ t3.set(3, 10);
+
+ DataBag bag = bf.newDefaultBag();
+ bag.add(t1);
+ bag.add(t2);
+ bag.add(t3);
+
+ Tuple udfInput = tf.newTuple(1);
+ udfInput.set(0, bag);
+
+ // invoking UDF
+ BagToTuple udf = new BagToTuple();
+ Tuple outputTuple = udf.exec(udfInput);
+
+ int totalExpectedSize = t1.size() + t2.size() + t3.size();
+ assertEquals(totalExpectedSize, outputTuple.size());
+
+ for (int i = 0; i < t1.size(); i++) {
+ assertEquals(t1.get(i), outputTuple.get(i));
+ }
+
+ for (int i = 0; i < t2.size(); i++) {
+ assertEquals(t2.get(i), outputTuple.get(t1.size() + i));
+ }
+
+ int startIndex = t1.size() + t2.size();
+ for (int i = 0; i < t3.size(); i++) {
+ assertEquals(t3.get(i), outputTuple.get(startIndex + i));
+ }
+ }
+
+ @Test
+ public void testNestedDataElementsForBagToTupleUDF() throws Exception {
+
+ DataBag inputBag = buildBagWithNestedTupleAndBag();
+
+
+ BagToTuple udf = new BagToTuple();
+ Tuple udfInput = tf.newTuple(1);
+ udfInput.set(0, inputBag);
+ Tuple outputTuple = udf.exec(udfInput);
+
+
+ Iterator<Tuple> inputBagIterator = inputBag.iterator();
+ Tuple firstTuple = inputBagIterator.next();
+ for (int i = 0; i < firstTuple.size(); i++) {
+ assertEquals(firstTuple.get(i), outputTuple.get(i));
+ }
+
+ Tuple secondTuple = inputBagIterator.next();
+ for (int i = 0; i < secondTuple.size(); i++) {
+ assertEquals(secondTuple.get(i), outputTuple.get(firstTuple.size() + i));
+ }
+
+ int startIndex = firstTuple.size() + secondTuple.size();
+ Tuple thirdTuple = inputBagIterator.next();
+ for (int i = 0; i < thirdTuple.size(); i++) {
+ assertEquals(thirdTuple.get(i), outputTuple.get(startIndex + i));
+ }
+ }
+
+ @Test
+ public void testOutputSchemaForBagToTupleUDF() throws Exception {
+ Schema expectedSch = Schema.generateNestedSchema(DataType.TUPLE,
+ DataType.INTEGER, DataType.CHARARRAY);
+
+ FieldSchema tupSch = new FieldSchema(null, DataType.TUPLE);
+ tupSch.schema = new Schema();
+ tupSch.schema.add(new FieldSchema(null, DataType.INTEGER));
+ tupSch.schema.add(new FieldSchema(null, DataType.CHARARRAY));
+
+ FieldSchema bagSch = new FieldSchema(null, DataType.BAG);
+ bagSch.schema = new Schema(tupSch);
+
+ Schema inputSch = new Schema();
+ inputSch.add(bagSch);
+
+ BagToTuple udf = new BagToTuple();
+ Schema outputSchema = udf.outputSchema(inputSch);
+
+ assertEquals("schema of BagToTuple input", expectedSch.size(),
+ outputSchema.size());
+ assertTrue("schema of BagToTuple input",
+ Schema.equals(expectedSch, outputSchema, false, true));
+ }
+
+ @Test(expected=org.apache.pig.backend.executionengine.ExecException.class)
+ public void testInvalidInputToBagToTupleUDF() throws Exception {
+ TupleFactory tf = TupleFactory.getInstance();
+ Tuple udfInput = tf.newTuple(1);
+ // input contains tuple instead of bag
+ udfInput.set(0, tf.newTuple());
+ BagToTuple udf = new BagToTuple();
+
+ // expecting an exception because the input if of type Tuple, not DataBag
+ udf.exec(udfInput);
+ }
+
+
+ @Test
+ public void testNullInputBagToStringUDF() throws Exception {
+ BagToString udf = new BagToString();
+ Tuple udfInput = tf.newTuple(1);
+ udfInput.set(0, null);
+ String output = udf.exec(udfInput);
+ assertNull(output);
+ }
+
+ @Test(expected=org.apache.pig.backend.executionengine.ExecException.class)
+ public void testInvalidInputForBagToStringUDF() throws Exception {
+ TupleFactory tf = TupleFactory.getInstance();
+ Tuple udfInput = tf.newTuple(1);
+ // input contains tuple instead of bag
+ udfInput.set(0, tf.newTuple());
+ BagToString udf = new BagToString();
+
+ // expecting an exception because the input if of type Tuple, not DataBag
+ udf.exec(udfInput);
+ }
+
+ @Test
+ public void testUseDefaultDelimiterBagToStringUDF() throws Exception {
+ BagFactory bf = BagFactory.getInstance();
+ TupleFactory tf = TupleFactory.getInstance();
+
+ Tuple t1 = tf.newTuple(2);
+ t1.set(0, "a");
+ t1.set(1, 5);
+
+ Tuple t2 = tf.newTuple(2);
+ t2.set(0, "c");
+ t2.set(1, 6);
+
+ DataBag bag = bf.newDefaultBag();
+ bag.add(t1);
+ bag.add(t2);
+
+ BagToString udf = new BagToString();
+ Tuple udfInput = tf.newTuple(1);
+ udfInput.set(0, bag);
+ String result = udf.exec(udfInput);
+
+ assertEquals("a_5_c_6", result);
+ }
+
+ @Test
+ public void testBasicBagToStringUDF() throws Exception {
+ BagFactory bf = BagFactory.getInstance();
+ TupleFactory tf = TupleFactory.getInstance();
+
+ Tuple t1 = tf.newTuple(2);
+ t1.set(0, "a");
+ t1.set(1, 5);
+
+ Tuple t2 = tf.newTuple(2);
+ t2.set(0, "c");
+ t2.set(1, 6);
+
+ DataBag bag = bf.newDefaultBag();
+ bag.add(t1);
+ bag.add(t2);
+
+ BagToString udf = new BagToString();
+ Tuple udfInput = tf.newTuple(2);
+ udfInput.set(0, bag);
+ udfInput.set(1, "-");
+ String result = udf.exec(udfInput);
+
+ assertEquals("a-5-c-6", result);
+ }
+
+ @Test
+ public void testNestedTupleForBagToStringUDF() throws Exception {
+ BagFactory bf = BagFactory.getInstance();
+ TupleFactory tf = TupleFactory.getInstance();
+
+ Tuple t1 = tf.newTuple(2);
+ t1.set(0, "a");
+ t1.set(1, 5);
+
+ Tuple nestedTuple = tf.newTuple(2);
+ nestedTuple.set(0, "d");
+ nestedTuple.set(1, 7);
+
+ Tuple t2 = tf.newTuple(3);
+ t2.set(0, "c");
+ t2.set(1, 6);
+ t2.set(2, nestedTuple);
+
+ DataBag inputBag = bf.newDefaultBag();
+ inputBag.add(t1);
+ inputBag.add(t2);
+
+ BagToString udf = new BagToString();
+ Tuple udfInput = tf.newTuple(2);
+ udfInput.set(0, inputBag);
+ udfInput.set(1, "_");
+ String result = udf.exec(udfInput);
+
+ assertEquals("a_5_c_6_(d,7)", result);
+ }
+
+ @Test
+ public void testNestedDataElementsForBagToStringUDF() throws Exception {
+
+ DataBag inputBag = buildBagWithNestedTupleAndBag();
+
+ BagToString udf = new BagToString();
+ Tuple udfInput = tf.newTuple(2);
+ udfInput.set(0, inputBag);
+ udfInput.set(1, "*");
+
+ String result = udf.exec(udfInput);
+ assertEquals("a*5*c*6*(d,7)*{(in bag,10)}", result);
+ }
+
+
+ @Test(expected=java.lang.RuntimeException.class)
+ public void testInvalidZeroInputToOutputSchemaForBagToTupleStringUDF() throws Exception {
+
+
+ Schema inputSch = new Schema();
+
+ BagToString udf = new BagToString();
+ Schema outputSchema = udf.outputSchema(inputSch);
+
+ assertEquals("schema of BagToTuple input", outputSchema.getField(0).type,
+ DataType.CHARARRAY);
+
+ }
+
+ @Test
+ public void testOutputSchemaForBagToTupleStringUDF() throws Exception {
+
+ FieldSchema tupSch = new FieldSchema(null, DataType.TUPLE);
+ tupSch.schema = new Schema();
+ tupSch.schema.add(new FieldSchema(null, DataType.INTEGER));
+ tupSch.schema.add(new FieldSchema(null, DataType.CHARARRAY));
+
+ FieldSchema bagSch = new FieldSchema(null, DataType.BAG);
+ bagSch.schema = new Schema(tupSch);
+
+ Schema inputSch = new Schema();
+ inputSch.add(bagSch);
+ inputSch.add(new FieldSchema(null, DataType.CHARARRAY));
+
+ BagToString udf = new BagToString();
+ Schema outputSchema = udf.outputSchema(inputSch);
+
+ assertEquals("schema of BagToTuple input", outputSchema.getField(0).type,
+ DataType.CHARARRAY);
+
+ }
+
+ @Test
+ public void testOutputSchemaWithDefaultDelimiterForBagToTupleStringUDF() throws Exception {
+
+ FieldSchema tupSch = new FieldSchema(null, DataType.TUPLE);
+ tupSch.schema = new Schema();
+ tupSch.schema.add(new FieldSchema(null, DataType.INTEGER));
+ tupSch.schema.add(new FieldSchema(null, DataType.CHARARRAY));
+
+ FieldSchema bagSch = new FieldSchema(null, DataType.BAG);
+ bagSch.schema = new Schema(tupSch);
+
+ Schema inputSch = new Schema();
+ inputSch.add(bagSch);
+
+ BagToString udf = new BagToString();
+ Schema outputSchema = udf.outputSchema(inputSch);
+
+ assertEquals("schema of BagToTuple input", outputSchema.getField(0).type,
+ DataType.CHARARRAY);
+
+ }
+
+ @Test(expected=java.lang.RuntimeException.class)
+ public void testInvalidOutputSchemaForBagToTupleStringUDF() throws Exception {
+
+ FieldSchema tupSch = new FieldSchema(null, DataType.TUPLE);
+ tupSch.schema = new Schema();
+ tupSch.schema.add(new FieldSchema(null, DataType.INTEGER));
+ tupSch.schema.add(new FieldSchema(null, DataType.CHARARRAY));
+
+ FieldSchema bagSch = new FieldSchema(null, DataType.BAG);
+ bagSch.schema = new Schema(tupSch);
+
+ Schema inputSch = new Schema();
+ inputSch.add(bagSch);
+ inputSch.add(new FieldSchema(null, DataType.DOUBLE));
+
+ BagToString udf = new BagToString();
+ // expecting an exception because the delimiter is not of type Data.CHARARRAY
+ udf.outputSchema(inputSch);
+ }
+
+ @Test
+ public void testPigScriptForBagToTupleUDF() throws Exception {
+ PigServer pigServer = new PigServer(ExecType.LOCAL);
+ Data data = resetData(pigServer);
+
+ // bag of chararray
+ data.set("foo", "myBag:bag{t:(l:chararray)}",
+ tuple(bag(tuple("a"), tuple("b"), tuple("c"))));
+ pigServer.registerQuery("A = LOAD 'foo' USING mock.Storage();");
+ pigServer.registerQuery("B = FOREACH A GENERATE BagToTuple(myBag) as myBag;");
+ pigServer.registerQuery("STORE B INTO 'bar' USING mock.Storage();");
+
+ assertEquals(schema("myBag:(l:chararray)"), data.getSchema("bar"));
+
+ List<Tuple> out = data.get("bar");
+ assertEquals(tuple("a", "b","c"), out.get(0).get(0));
+
+ // bag of longs
+ data = resetData(pigServer);
+ data.set("foo", "myBag:bag{t:(l:long)}",
+ tuple(bag(tuple(1), tuple(2), tuple(3))));
+ pigServer.registerQuery("A = LOAD 'foo' USING mock.Storage();");
+ pigServer.registerQuery("B = FOREACH A GENERATE BagToTuple(myBag) as myBag;");
+ pigServer.registerQuery("STORE B INTO 'bar' USING mock.Storage();");
+
+ out = data.get("bar");
+ assertEquals(tuple(1, 2, 3), out.get(0).get(0));
+ }
+
+ @Test
+ public void testPigScriptMultipleElmementsPerTupleForBagTupleUDF() throws Exception {
+ PigServer pigServer = new PigServer(ExecType.LOCAL);
+ Data data = resetData(pigServer);
+
+ data.set("foo", "myBag:bag{t:(l:chararray)}",
+ tuple(bag(tuple("a", "b"), tuple("c", "d"), tuple("e", "f"))));
+ pigServer.registerQuery("A = LOAD 'foo' USING mock.Storage();");
+ pigServer.registerQuery("B = FOREACH A GENERATE BagToTuple(myBag) as myBag;");
+ pigServer.registerQuery("STORE B INTO 'bar' USING mock.Storage();");
+
+ List<Tuple> out = data.get("bar");
+ assertEquals(tuple("a", "b","c", "d", "e", "f"), out.get(0).get(0));
+ }
+
+ @Test
+ public void testPigScriptNestedTupleForBagToTupleDF() throws Exception {
+ PigServer pigServer = new PigServer(ExecType.LOCAL);
+ Data data = resetData(pigServer);
+
+ Tuple nestedTuple = tuple(bag(tuple("c"), tuple("d")));
+ data.set("foo", "myBag:bag{t:(l:chararray)}",
+ tuple(bag(tuple("a"), tuple("b"), nestedTuple, tuple("e"))));
+
+ pigServer.registerQuery("A = LOAD 'foo' USING mock.Storage();");
+ pigServer.registerQuery("B = FOREACH A GENERATE BagToTuple(myBag) as myBag;");
+ pigServer.registerQuery("STORE B INTO 'bar' USING mock.Storage();");
+
+ List<Tuple> out = data.get("bar");
+ assertEquals(tuple("a", "b",bag(tuple("c"), tuple("d")), "e"), out.get(0).get(0));
+
+ }
+
+ @Test
+ public void testPigScriptEmptyBagForBagToTupleUDF() throws Exception {
+ PigServer pigServer = new PigServer(ExecType.LOCAL);
+ Data data = resetData(pigServer);
+
+ data.set("foo", "myBag:bag{t:(l:chararray)}",
+ tuple(bag()));
+
+ pigServer.registerQuery("A = LOAD 'foo' USING mock.Storage();");
+ pigServer.registerQuery("B = FOREACH A GENERATE BagToTuple(myBag) as myBag;");
+ pigServer.registerQuery("STORE B INTO 'bar' USING mock.Storage();");
+
+ List<Tuple> out = data.get("bar");
+ // empty bag will generate empty tuple
+ assertEquals(tuple(), out.get(0).get(0));
+
+ }
+
+ @Test
+ public void testPigScriptrForBagToStringUDF() throws Exception {
+ PigServer pigServer = new PigServer(ExecType.LOCAL);
+ Data data = resetData(pigServer);
+
+ data.set("foo", "myBag:bag{t:(l:chararray)}",
+ tuple(bag(tuple("a"), tuple("b"), tuple("c"))));
+ pigServer.registerQuery("A = LOAD 'foo' USING mock.Storage();");
+ pigServer.registerQuery("B = FOREACH A GENERATE BagToString(myBag) as myBag;");
+ pigServer.registerQuery("STORE B INTO 'bar' USING mock.Storage();");
+
+ pigServer.registerQuery("C = FOREACH A GENERATE BagToString(myBag, '==') as myBag;");
+ pigServer.registerQuery("STORE C INTO 'foo' USING mock.Storage();");
+
+ List<Tuple> out = data.get("bar");
+ assertEquals(schema("myBag:chararray"), data.getSchema("bar"));
+ assertEquals(tuple("a_b_c"), out.get(0));
+
+ out = data.get("foo");
+ assertEquals(tuple("a==b==c"), out.get(0));
+ }
+
+ @Test
+ public void testPigScriptMultipleElmementsPerTupleForBagToStringUDF() throws Exception {
+ PigServer pigServer = new PigServer(ExecType.LOCAL);
+ Data data = resetData(pigServer);
+
+ data.set("foo", "myBag:bag{t:(l:chararray)}",
+ tuple(bag(tuple("a", "b"), tuple("c", "d"), tuple("e", "f"))));
+ pigServer.registerQuery("A = LOAD 'foo' USING mock.Storage();");
+ pigServer.registerQuery("B = FOREACH A GENERATE BagToString(myBag) as myBag;");
+ pigServer.registerQuery("STORE B INTO 'bar' USING mock.Storage();");
+
+ pigServer.registerQuery("C = FOREACH A GENERATE BagToString(myBag, '^') as myBag;");
+ pigServer.registerQuery("STORE C INTO 'foo' USING mock.Storage();");
+
+ List<Tuple> out = data.get("bar");
+ assertEquals(tuple("a_b_c_d_e_f"), out.get(0));
+
+ out = data.get("foo");
+ assertEquals(tuple("a^b^c^d^e^f"), out.get(0));
+ }
+
+ @Test
+ public void testPigScriptNestedTupleForBagToStringUDF() throws Exception {
+ PigServer pigServer = new PigServer(ExecType.LOCAL);
+ Data data = resetData(pigServer);
+
+ Tuple nestedTuple = tuple(bag(tuple("c"), tuple("d")));
+ data.set("foo", "myBag:bag{t:(l:chararray)}",
+ tuple(bag(tuple("a"), tuple("b"), nestedTuple, tuple("e"))));
+
+ pigServer.registerQuery("A = LOAD 'foo' USING mock.Storage();");
+ pigServer.registerQuery("B = FOREACH A GENERATE BagToString(myBag) as myBag;");
+ pigServer.registerQuery("STORE B INTO 'bar' USING mock.Storage();");
+
+ List<Tuple> out = data.get("bar");
+ assertEquals(tuple("a_b_{(c),(d)}_e"), out.get(0));
+
+ }
+
+ @Test
+ public void testPigScriptEmptyBagForBagToStringUDF() throws Exception {
+ PigServer pigServer = new PigServer(ExecType.LOCAL);
+ Data data = resetData(pigServer);
+
+ data.set("foo", "myBag:bag{t:(l:chararray)}",
+ tuple(bag()));
+
+ pigServer.registerQuery("A = LOAD 'foo' USING mock.Storage();");
+ pigServer.registerQuery("B = FOREACH A GENERATE BagToString(myBag) as myBag;");
+ pigServer.registerQuery("STORE B INTO 'bar' USING mock.Storage();");
+
+ List<Tuple> out = data.get("bar");
+ // empty bag will generate empty string
+ assertEquals(tuple(""), out.get(0));
+
+ }
+
+ private DataBag buildBagWithNestedTupleAndBag() throws ExecException {
+ Tuple t1 = tf.newTuple(2);
+ t1.set(0, "a");
+ t1.set(1, 5);
+
+ Tuple nestedTuple = tf.newTuple(2);
+ nestedTuple.set(0, "d");
+ nestedTuple.set(1, 7);
+
+ Tuple t2 = tf.newTuple(3);
+ t2.set(0, "c");
+ t2.set(1, 6);
+ t2.set(2, nestedTuple);
+
+ DataBag nestedBag = bf.newDefaultBag();
+ Tuple tupleInNestedBag = tf.newTuple(2);
+ tupleInNestedBag.set(0, "in bag");
+ tupleInNestedBag.set(1, 10);
+ nestedBag.add(tupleInNestedBag);
+
+ Tuple t3 = tf.newTuple(1);
+ t3.set(0, nestedBag);
+
+ DataBag bag = bf.newDefaultBag();
+ bag.add(t1);
+ bag.add(t2);
+ bag.add(t3);
+ return bag;
+ }
+}