You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by jc...@apache.org on 2013/01/23 00:30:00 UTC
svn commit: r1437235 - in /pig/trunk: CHANGES.txt
src/org/apache/pig/builtin/PluckTuple.java
test/org/apache/pig/builtin/TestPluckTuple.java
Author: jcoveney
Date: Tue Jan 22 23:30:00 2013
New Revision: 1437235
URL: http://svn.apache.org/viewvc?rev=1437235&view=rev
Log:
PIG-3078: Make a UDF that, given a string, returns just the columns prefixed by that string (jcoveney)
Added:
pig/trunk/src/org/apache/pig/builtin/PluckTuple.java
pig/trunk/test/org/apache/pig/builtin/TestPluckTuple.java
Modified:
pig/trunk/CHANGES.txt
Modified: pig/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/pig/trunk/CHANGES.txt?rev=1437235&r1=1437234&r2=1437235&view=diff
==============================================================================
--- pig/trunk/CHANGES.txt (original)
+++ pig/trunk/CHANGES.txt Tue Jan 22 23:30:00 2013
@@ -24,6 +24,8 @@ INCOMPATIBLE CHANGES
IMPROVEMENTS
+PIG-3078: Make a UDF that, given a string, returns just the columns prefixed by that string (jcoveney)
+
PIG-3090: Introduce a syntax to be able to easily refer to the previously defined relation (jcoveney)
PIG-3057: Make PigStorage.readField() protected (pablomar and billgraham via billgraham)
Added: pig/trunk/src/org/apache/pig/builtin/PluckTuple.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/builtin/PluckTuple.java?rev=1437235&view=auto
==============================================================================
--- pig/trunk/src/org/apache/pig/builtin/PluckTuple.java (added)
+++ pig/trunk/src/org/apache/pig/builtin/PluckTuple.java Tue Jan 22 23:30:00 2013
@@ -0,0 +1,102 @@
+package org.apache.pig.builtin;
+
+import java.io.IOException;
+import java.util.List;
+
+import org.apache.pig.EvalFunc;
+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;
+
+import com.google.common.collect.Lists;
+
+/**
+ * This is a UDF which allows the user to specify a string prefix, and then
+ * filter for the columns in a relation that begin with that prefix.
+ *
+ * Example:
+ * a = load 'a' as (x, y);
+ * b = load 'b' as (x, y);
+ * c = join a by x, b by x;
+ * DEFINE pluck PluckTuple('a::');
+ * d = foreach c generate FLATTEN(pluck(*));
+ * describe c;
+ * c: {a::x: bytearray,a::y: bytearray,b::x: bytearray,b::y: bytearray}
+ * describe d;
+ * d: {plucked::a::x: bytearray,plucked::a::y: bytearray}
+ */
+public class PluckTuple extends EvalFunc<Tuple> {
+ private static final TupleFactory mTupleFactory = TupleFactory.getInstance();
+
+ private boolean isInitialized = false;
+ private int[] indicesToInclude;
+ private String prefix;
+
+ public PluckTuple(String prefix) {
+ this.prefix = prefix;
+ }
+
+ @Override
+ public Tuple exec(Tuple input) throws IOException {
+ if (!isInitialized) {
+ List<Integer> indicesToInclude = Lists.newArrayList();
+ Schema inputSchema = getInputSchema();
+ for (int i = 0; i < inputSchema.size(); i++) {
+ String alias = inputSchema.getField(i).alias;
+ if (alias.startsWith(prefix)) {
+ indicesToInclude.add(i);
+ }
+ }
+ this.indicesToInclude = new int[indicesToInclude.size()];
+ int idx = 0;
+ for (int val : indicesToInclude) {
+ this.indicesToInclude[idx++] = val;
+ }
+ isInitialized = true;
+ }
+ Tuple result = mTupleFactory.newTuple(indicesToInclude.length);
+ int idx = 0;
+ for (int val : indicesToInclude) {
+ result.set(idx++, input.get(val));
+ }
+ return result;
+ }
+
+ public Schema outputSchema(Schema inputSchema) {
+ if (!isInitialized) {
+ List<Integer> indicesToInclude = Lists.newArrayList();
+ for (int i = 0; i < inputSchema.size(); i++) {
+ String alias;
+ try {
+ alias = inputSchema.getField(i).alias;
+ } catch (FrontendException e) {
+ throw new RuntimeException(e); // Should never happen
+ }
+ if (alias.startsWith(prefix)) {
+ indicesToInclude.add(i);
+ }
+ }
+ this.indicesToInclude = new int[indicesToInclude.size()];
+ int idx = 0;
+ for (int val : indicesToInclude) {
+ this.indicesToInclude[idx++] = val;
+ }
+ isInitialized = true;
+ }
+ Schema outputSchema = new Schema();
+ for (int val : indicesToInclude) {
+ try {
+ outputSchema.add(inputSchema.getField(val));
+ } catch (FrontendException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ try {
+ return new Schema(new Schema.FieldSchema("plucked", outputSchema, DataType.TUPLE));
+ } catch (FrontendException e) {
+ throw new RuntimeException(e); // Should never happen
+ }
+ }
+}
\ No newline at end of file
Added: pig/trunk/test/org/apache/pig/builtin/TestPluckTuple.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/builtin/TestPluckTuple.java?rev=1437235&view=auto
==============================================================================
--- pig/trunk/test/org/apache/pig/builtin/TestPluckTuple.java (added)
+++ pig/trunk/test/org/apache/pig/builtin/TestPluckTuple.java Tue Jan 22 23:30:00 2013
@@ -0,0 +1,72 @@
+package org.apache.pig.builtin;
+
+import static org.apache.pig.builtin.mock.Storage.resetData;
+import static org.apache.pig.builtin.mock.Storage.tuple;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import java.util.Iterator;
+
+import org.apache.pig.ExecType;
+import org.apache.pig.PigServer;
+import org.apache.pig.builtin.mock.Storage.Data;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.impl.logicalLayer.schema.Schema;
+import org.apache.pig.impl.util.Utils;
+import org.junit.Before;
+import org.junit.Test;
+
+public class TestPluckTuple {
+ private PigServer pigServer;
+
+ @Before
+ public void setUp() throws Exception {
+ pigServer = new PigServer(ExecType.LOCAL);
+ }
+
+ @Test
+ public void testSchema() throws Exception {
+ String query = "a = load 'a' as (x:int,y:chararray,z:long);" +
+ "b = load 'b' as (x:int,y:chararray,z:long);" +
+ "c = join a by x, b by x;" +
+ "define pluck PluckTuple('a::');" +
+ "d = foreach c generate flatten(pluck(*));";
+ pigServer.registerQuery(query);
+ assertTrue(Schema.equals(pigServer.dumpSchema("a"), pigServer.dumpSchema("d"), false, true));
+ }
+
+ @Test
+ public void testOutput() throws Exception {
+ Data data = resetData(pigServer);
+
+ Tuple exp1 = tuple(1, "hey", 2L);
+ Tuple exp2 = tuple(2, "woah", 3L);
+
+ data.set("a",
+ Utils.getSchemaFromString("x:int,y:chararray,z:long"),
+ exp1,
+ exp2,
+ tuple(3, "c", 4L)
+ );
+ data.set("b",
+ Utils.getSchemaFromString("x:int,y:chararray,z:long"),
+ tuple(1, "sasf", 5L),
+ tuple(2, "woah", 6L),
+ tuple(4, "c", 7L)
+ );
+
+ String query = "a = load 'a' using mock.Storage();" +
+ "b = load 'b' using mock.Storage();" +
+ "c = join a by x, b by x;" +
+ "define pluck PluckTuple('a::');" +
+ "d = foreach c generate flatten(pluck(*));";
+ pigServer.registerQuery(query);
+ Iterator<Tuple> it = pigServer.openIterator("d");
+ assertTrue(it.hasNext());
+ assertEquals(exp1, it.next());
+ assertTrue(it.hasNext());
+ assertEquals(exp2, it.next());
+ assertFalse(it.hasNext());
+ }
+}
\ No newline at end of file