You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by ga...@apache.org on 2008/09/23 02:31:28 UTC
svn commit: r698044 - in /incubator/pig/branches/types: lib/
src/org/apache/pig/
src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/
src/org/apache/pig/impl/builtin/ test/org/apache/pig/test/
Author: gates
Date: Mon Sep 22 17:31:27 2008
New Revision: 698044
URL: http://svn.apache.org/viewvc?rev=698044&view=rev
Log:
PIG-441 Added object comparators to the PigXRawComparator classes so that in places where hadoop uses the object instead of null comparators the desc behavior
can still be handled.
Changed FindQuantiles UDF to take in its constructor an array that indicates ascending vs descending order so that the quantiles can be correclty computed.
Removed:
incubator/pig/branches/types/lib/hadoop16.jar
Modified:
incubator/pig/branches/types/src/org/apache/pig/FuncSpec.java
incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java
incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRCompiler.java
incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigBytesRawComparator.java
incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigDoubleRawComparator.java
incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigFloatRawComparator.java
incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigIntRawComparator.java
incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigLongRawComparator.java
incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigTextRawComparator.java
incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigTupleRawComparator.java
incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/SortPartitioner.java
incubator/pig/branches/types/src/org/apache/pig/impl/builtin/FindQuantiles.java
incubator/pig/branches/types/test/org/apache/pig/test/TestEvalPipeline.java
incubator/pig/branches/types/test/org/apache/pig/test/TestMRCompiler.java
Modified: incubator/pig/branches/types/src/org/apache/pig/FuncSpec.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/FuncSpec.java?rev=698044&r1=698043&r2=698044&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/FuncSpec.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/FuncSpec.java Mon Sep 22 17:31:27 2008
@@ -29,9 +29,6 @@
* Class to represent a UDF specification - essentially
* encapsulates the class name and the arguments to the constructor
*/
-/**
- *
- */
public class FuncSpec implements Serializable, Cloneable {
private static final long serialVersionUID = 2L;
Modified: incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java?rev=698044&r1=698043&r2=698044&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java Mon Sep 22 17:31:27 2008
@@ -454,34 +454,6 @@
}
}
- /*
- public static class PigRawGrouper extends WritableComparator {
- protected PigRawGrouper(Class c) {
- super(c);
- }
-
- public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2){
- // If it's null, then look at the index. Else, ignore the index.
- if (b1[s1] == 0 && b2[s2] == 0) {
- return WritableComparator.compareBytes(b1, s1 + 1, l1 - 2, b2, s2 + 1, l2 - 2);
- } else if (b1[s1] != 0 && b2[s2] != 0) {
- if (b1[s1 + l1 - 1] < b2[s2 + l2 - 1]) return -1;
- else if (b2[s2 + l2 - 1] > b1[s1 + l1 - 1]) return 1;
- else return 0;
- }
- else if (b1[s1] != 0) return -1;
- else return 1;
- }
- }
-
- public static class PigCharArrayRawGrouper extends PigRawGrouper {
- public PigCharArrayRawGrouper() {
- super(NullableText.class);
- }
- }
- */
-
-
private void selectComparator(
MapReduceOper mro,
byte keyType,
@@ -548,72 +520,6 @@
return;
}
- /*
- try {
- CogroupFinder cf = new CogroupFinder(mro.mapPlan);
- cf.visit();
- int mapRearranges = cf.rearrangeCounter;
- cf = new CogroupFinder(mro.reducePlan);
- cf.visit();
- if (mapRearranges > 1 || cf.rearrangeCounter > 1) {
- switch (keyType) {
- case DataType.INTEGER:
- jobConf.setOutputKeyComparatorClass(PigIntWritableComparator.class);
- jobConf.setOutputValueGroupingComparator(PigIntRawComparator.class);
- break;
-
- case DataType.LONG:
- jobConf.setOutputKeyComparatorClass(PigLongWritableComparator.class);
- jobConf.setOutputValueGroupingComparator(PigLongRawComparator.class);
- break;
-
- case DataType.FLOAT:
- jobConf.setOutputKeyComparatorClass(PigFloatWritableComparator.class);
- jobConf.setOutputValueGroupingComparator(PigFloatRawComparator.class);
- break;
-
- case DataType.DOUBLE:
- jobConf.setOutputKeyComparatorClass(PigDoubleWritableComparator.class);
- jobConf.setOutputValueGroupingComparator(PigDoubleRawComparator.class);
- break;
-
- case DataType.CHARARRAY:
- jobConf.setOutputKeyComparatorClass(PigCharArrayWritableComparator.class);
- jobConf.setOutputValueGroupingComparator(PigCharArrayRawGrouper.class);
- break;
-
- case DataType.BYTEARRAY:
- jobConf.setOutputKeyComparatorClass(PigDBAWritableComparator.class);
- jobConf.setOutputValueGroupingComparator(PigBytesRawComparator.class);
- break;
-
- case DataType.MAP:
- log.error("Using Map as key not supported.");
- throw new JobCreationException("Using Map as key not supported");
-
- case DataType.TUPLE:
- jobConf.setOutputKeyComparatorClass(PigTupleWritableComparator.class);
- jobConf.setOutputValueGroupingComparator(PigTupleRawComparator.class);
- break;
-
- case DataType.BAG:
- log.error("Using Bag as key not supported.");
- throw new JobCreationException("Using Bag as key not supported");
-
- default:
- throw new RuntimeException("Forgot case for type " +
- DataType.findTypeName(keyType));
- }
- */
- jobConf.setPartitionerClass(org.apache.hadoop.mapred.lib.HashPartitioner.class);
- /*
- return;
- }
- } catch (VisitorException ve) {
- throw new JobCreationException(ve);
- }
- */
-
switch (keyType) {
case DataType.INTEGER:
jobConf.setOutputKeyComparatorClass(PigIntWritableComparator.class);
@@ -708,20 +614,4 @@
}
}
}
-
- /*
- private class CogroupFinder extends PhyPlanVisitor {
- int rearrangeCounter = 0;
-
- CogroupFinder(PhysicalPlan plan) {
- super(plan,
- new DepthFirstWalker<PhysicalOperator, PhysicalPlan>(plan));
- }
-
- public void visitLocalRearrange(POLocalRearrange lr) {
- rearrangeCounter++;
- }
- }
- */
-
}
Modified: incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRCompiler.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRCompiler.java?rev=698044&r1=698043&r2=698044&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRCompiler.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRCompiler.java Mon Sep 22 17:31:27 2008
@@ -1144,8 +1144,13 @@
List ufInps = new ArrayList();
ufInps.add(prjStar4);
+ // Turn the asc/desc array into an array of strings so that we can pass it
+ // to the FindQuantiles function.
+ List<Boolean> ascCols = inpSort.getMAscCols();
+ String[] ascs = new String[ascCols.size()];
+ for (int i = 0; i < ascCols.size(); i++) ascs[i] = ascCols.get(i).toString();
POUserFunc uf = new POUserFunc(new OperatorKey(scope,nig.getNextNodeId(scope)), -1, ufInps,
- new FuncSpec(FindQuantiles.class.getName()));
+ new FuncSpec(FindQuantiles.class.getName(), ascs));
ep4.add(uf);
ep4.connect(prjStar4, uf);
Modified: incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigBytesRawComparator.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigBytesRawComparator.java?rev=698044&r1=698043&r2=698044&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigBytesRawComparator.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigBytesRawComparator.java Mon Sep 22 17:31:27 2008
@@ -25,17 +25,23 @@
import org.apache.hadoop.conf.Configurable;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.BytesWritable;
-import org.apache.hadoop.io.RawComparator;
+import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapred.JobConf;
+import org.apache.pig.data.DataByteArray;
import org.apache.pig.impl.io.NullableBytesWritable;
-import org.apache.pig.impl.io.PigNullableWritable;
import org.apache.pig.impl.util.ObjectSerializer;
-public class PigBytesRawComparator extends BytesWritable.Comparator implements Configurable {
+public class PigBytesRawComparator extends WritableComparator implements Configurable {
private final Log mLog = LogFactory.getLog(getClass());
private boolean[] mAsc;
+ private BytesWritable.Comparator mWrappedComp;
+
+ public PigBytesRawComparator() {
+ super(NullableBytesWritable.class);
+ mWrappedComp = new BytesWritable.Comparator();
+ }
public void setConf(Configuration conf) {
if (!(conf instanceof JobConf)) {
@@ -73,7 +79,7 @@
// If either are null, handle differently.
if (b1[s1] == 0 && b2[s2] == 0) {
// Subtract 2, one for null byte and one for index byte
- rc = super.compare(b1, s1 + 1, l1 - 2, b2, s2 + 1, l2 - 2);
+ rc = mWrappedComp.compare(b1, s1 + 1, l1 - 2, b2, s2 + 1, l2 - 2);
} else {
// For sorting purposes two nulls are equal.
if (b1[s1] != 0 && b2[s2] != 0) rc = 0;
@@ -84,5 +90,22 @@
return rc;
}
+ public int compare(Object o1, Object o2) {
+ NullableBytesWritable nbw1 = (NullableBytesWritable)o1;
+ NullableBytesWritable nbw2 = (NullableBytesWritable)o2;
+ int rc = 0;
+
+ // If either are null, handle differently.
+ if (!nbw1.isNull() && !nbw2.isNull()) {
+ rc = ((DataByteArray)nbw1.getValueAsPigType()).compareTo((DataByteArray)nbw2.getValueAsPigType());
+ } else {
+ // For sorting purposes two nulls are equal.
+ if (nbw1.isNull() && nbw2.isNull()) rc = 0;
+ else if (nbw1.isNull()) rc = -1;
+ else rc = 1;
+ }
+ if (!mAsc[0]) rc *= -1;
+ return rc;
+ }
}
Modified: incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigDoubleRawComparator.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigDoubleRawComparator.java?rev=698044&r1=698043&r2=698044&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigDoubleRawComparator.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigDoubleRawComparator.java Mon Sep 22 17:31:27 2008
@@ -24,17 +24,23 @@
import org.apache.hadoop.conf.Configurable;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapred.JobConf;
import org.apache.pig.backend.hadoop.DoubleWritable;
import org.apache.pig.impl.io.NullableDoubleWritable;
-import org.apache.pig.impl.io.PigNullableWritable;
import org.apache.pig.impl.util.ObjectSerializer;
-public class PigDoubleRawComparator extends DoubleWritable.Comparator implements Configurable {
+public class PigDoubleRawComparator extends WritableComparator implements Configurable {
private final Log mLog = LogFactory.getLog(getClass());
private boolean[] mAsc;
+ private DoubleWritable.Comparator mWrappedComp;
+
+ public PigDoubleRawComparator() {
+ super(NullableDoubleWritable.class);
+ mWrappedComp = new DoubleWritable.Comparator();
+ }
public void setConf(Configuration conf) {
if (!(conf instanceof JobConf)) {
@@ -71,7 +77,7 @@
// If either are null, handle differently.
if (b1[s1] == 0 && b2[s2] == 0) {
- rc = super.compare(b1, s1 + 1, l1 - 2, b2, s2 + 1, l2 - 2);
+ rc = mWrappedComp.compare(b1, s1 + 1, l1 - 2, b2, s2 + 1, l2 - 2);
} else {
// For sorting purposes two nulls are equal.
if (b1[s1] != 0 && b2[s2] != 0) rc = 0;
@@ -82,5 +88,21 @@
return rc;
}
+ public int compare(Object o1, Object o2) {
+ NullableDoubleWritable ndw1 = (NullableDoubleWritable)o1;
+ NullableDoubleWritable ndw2 = (NullableDoubleWritable)o2;
+ int rc = 0;
+ // If either are null, handle differently.
+ if (!ndw1.isNull() && !ndw2.isNull()) {
+ rc = ((Double)ndw1.getValueAsPigType()).compareTo((Double)ndw2.getValueAsPigType());
+ } else {
+ // For sorting purposes two nulls are equal.
+ if (ndw1.isNull() && ndw2.isNull()) rc = 0;
+ else if (ndw1.isNull()) rc = -1;
+ else rc = 1;
+ }
+ if (!mAsc[0]) rc *= -1;
+ return rc;
+ }
}
Modified: incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigFloatRawComparator.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigFloatRawComparator.java?rev=698044&r1=698043&r2=698044&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigFloatRawComparator.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigFloatRawComparator.java Mon Sep 22 17:31:27 2008
@@ -25,17 +25,22 @@
import org.apache.hadoop.conf.Configurable;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.FloatWritable;
-import org.apache.hadoop.io.RawComparator;
+import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapred.JobConf;
import org.apache.pig.impl.io.NullableFloatWritable;
-import org.apache.pig.impl.io.PigNullableWritable;
import org.apache.pig.impl.util.ObjectSerializer;
-public class PigFloatRawComparator extends FloatWritable.Comparator implements Configurable {
+public class PigFloatRawComparator extends WritableComparator implements Configurable {
private final Log mLog = LogFactory.getLog(getClass());
private boolean[] mAsc;
+ private FloatWritable.Comparator mWrappedComp;
+
+ public PigFloatRawComparator() {
+ super(NullableFloatWritable.class);
+ mWrappedComp = new FloatWritable.Comparator();
+ }
public void setConf(Configuration conf) {
if (!(conf instanceof JobConf)) {
@@ -72,7 +77,7 @@
// If either are null, handle differently.
if (b1[s1] == 0 && b2[s2] == 0) {
- rc = super.compare(b1, s1 + 1, l1 - 2, b2, s2 + 1, l2 - 2);
+ rc = mWrappedComp.compare(b1, s1 + 1, l1 - 2, b2, s2 + 1, l2 - 2);
} else {
// For sorting purposes two nulls are equal.
if (b1[s1] != 0 && b2[s2] != 0) rc = 0;
@@ -83,5 +88,24 @@
return rc;
}
+ public int compare(Object o1, Object o2) {
+ NullableFloatWritable nfw1 = (NullableFloatWritable)o1;
+ NullableFloatWritable nfw2 = (NullableFloatWritable)o2;
+ int rc = 0;
+
+ // If either are null, handle differently.
+ if (!nfw1.isNull() && !nfw2.isNull()) {
+ rc = ((Float)nfw1.getValueAsPigType()).compareTo((Float)nfw2.getValueAsPigType());
+ } else {
+ // For sorting purposes two nulls are equal.
+ if (nfw1.isNull() && nfw2.isNull()) rc = 0;
+ else if (nfw1.isNull()) rc = -1;
+ else rc = 1;
+ }
+ if (!mAsc[0]) rc *= -1;
+ return rc;
+ }
+
+
}
Modified: incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigIntRawComparator.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigIntRawComparator.java?rev=698044&r1=698043&r2=698044&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigIntRawComparator.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigIntRawComparator.java Mon Sep 22 17:31:27 2008
@@ -24,18 +24,21 @@
import org.apache.hadoop.conf.Configurable;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.io.IntWritable;
-import org.apache.hadoop.io.RawComparator;
+import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapred.JobConf;
-import org.apache.pig.impl.io.PigNullableWritable;
+import org.apache.pig.impl.io.NullableIntWritable;
import org.apache.pig.impl.util.ObjectSerializer;
-public class PigIntRawComparator extends IntWritable.Comparator implements Configurable {
+public class PigIntRawComparator extends WritableComparator implements Configurable {
private final Log mLog = LogFactory.getLog(getClass());
private boolean[] mAsc;
+ public PigIntRawComparator() {
+ super(NullableIntWritable.class);
+ }
+
public void setConf(Configuration conf) {
if (!(conf instanceof JobConf)) {
mLog.warn("Expected jobconf in setConf, got " +
@@ -71,7 +74,9 @@
// If either are null, handle differently.
if (b1[s1] == 0 && b2[s2] == 0) {
- rc = super.compare(b1, s1 + 1, l1 - 2, b2, s2 + 1, l2 - 2);
+ int int1 = readInt(b1, s1 + 1);
+ int int2 = readInt(b2, s2 + 1);
+ rc = (int1 < int2) ? -1 : ((int1 > int2) ? 1 : 0);
} else {
// For sorting purposes two nulls are equal.
if (b1[s1] != 0 && b2[s2] != 0) rc = 0;
@@ -82,5 +87,23 @@
return rc;
}
+ public int compare(Object o1, Object o2) {
+ NullableIntWritable niw1 = (NullableIntWritable)o1;
+ NullableIntWritable niw2 = (NullableIntWritable)o2;
+ int rc = 0;
+
+ // If either are null, handle differently.
+ if (!niw1.isNull() && !niw2.isNull()) {
+ rc = ((Integer)niw1.getValueAsPigType()).compareTo((Integer)niw2.getValueAsPigType());
+ } else {
+ // For sorting purposes two nulls are equal.
+ if (niw1.isNull() && niw2.isNull()) rc = 0;
+ else if (niw1.isNull()) rc = -1;
+ else rc = 1;
+ }
+ if (!mAsc[0]) rc *= -1;
+ return rc;
+ }
+
}
Modified: incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigLongRawComparator.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigLongRawComparator.java?rev=698044&r1=698043&r2=698044&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigLongRawComparator.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigLongRawComparator.java Mon Sep 22 17:31:27 2008
@@ -25,17 +25,24 @@
import org.apache.hadoop.conf.Configurable;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
-import org.apache.hadoop.io.RawComparator;
+import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapred.JobConf;
import org.apache.pig.impl.io.NullableLongWritable;
-import org.apache.pig.impl.io.PigNullableWritable;
+import org.apache.pig.impl.io.NullableLongWritable;
import org.apache.pig.impl.util.ObjectSerializer;
-public class PigLongRawComparator extends LongWritable.Comparator implements Configurable {
+public class PigLongRawComparator extends WritableComparator implements Configurable {
private final Log mLog = LogFactory.getLog(getClass());
private boolean[] mAsc;
+ private LongWritable.Comparator mWrappedComp;
+
+ public PigLongRawComparator() {
+ super(NullableLongWritable.class);
+ mWrappedComp = new LongWritable.Comparator();
+ }
+
public void setConf(Configuration conf) {
if (!(conf instanceof JobConf)) {
@@ -72,7 +79,7 @@
// If either are null, handle differently.
if (b1[s1] == 0 && b2[s2] == 0) {
- rc = super.compare(b1, s1 + 1, l1 - 2, b2, s2 + 1, l2 - 2);
+ rc = mWrappedComp.compare(b1, s1 + 1, l1 - 2, b2, s2 + 1, l2 - 2);
} else {
// For sorting purposes two nulls are equal.
if (b1[s1] != 0 && b2[s2] != 0) rc = 0;
@@ -83,5 +90,23 @@
return rc;
}
+ public int compare(Object o1, Object o2) {
+ NullableLongWritable nlw1 = (NullableLongWritable)o1;
+ NullableLongWritable nlw2 = (NullableLongWritable)o2;
+ int rc = 0;
+
+ // If either are null, handle differently.
+ if (!nlw1.isNull() && !nlw2.isNull()) {
+ rc = ((Long)nlw1.getValueAsPigType()).compareTo((Long)nlw2.getValueAsPigType());
+ } else {
+ // For sorting purposes two nulls are equal.
+ if (nlw1.isNull() && nlw2.isNull()) rc = 0;
+ else if (nlw1.isNull()) rc = -1;
+ else rc = 1;
+ }
+ if (!mAsc[0]) rc *= -1;
+ return rc;
+ }
+
}
Modified: incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigTextRawComparator.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigTextRawComparator.java?rev=698044&r1=698043&r2=698044&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigTextRawComparator.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigTextRawComparator.java Mon Sep 22 17:31:27 2008
@@ -25,17 +25,23 @@
import org.apache.hadoop.conf.Configurable;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Text;
-import org.apache.hadoop.io.RawComparator;
+import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapred.JobConf;
import org.apache.pig.impl.io.NullableText;
-import org.apache.pig.impl.io.PigNullableWritable;
import org.apache.pig.impl.util.ObjectSerializer;
-public class PigTextRawComparator extends Text.Comparator implements Configurable {
+public class PigTextRawComparator extends WritableComparator implements Configurable {
private final Log mLog = LogFactory.getLog(getClass());
private boolean[] mAsc;
+ private Text.Comparator mWrappedComp;
+
+ public PigTextRawComparator() {
+ super(NullableText.class);
+ mWrappedComp = new Text.Comparator();
+ }
+
public void setConf(Configuration conf) {
if (!(conf instanceof JobConf)) {
@@ -72,7 +78,7 @@
// If either are null, handle differently.
if (b1[s1] == 0 && b2[s2] == 0) {
- rc = super.compare(b1, s1 + 1, l1 - 2, b2, s2 + 1, l2 - 2);
+ rc = mWrappedComp.compare(b1, s1 + 1, l1 - 2, b2, s2 + 1, l2 - 2);
} else {
// For sorting purposes two nulls are equal.
if (b1[s1] != 0 && b2[s2] != 0) rc = 0;
@@ -83,5 +89,24 @@
return rc;
}
+ public int compare(Object o1, Object o2) {
+ NullableText nt1 = (NullableText)o1;
+ NullableText nt2 = (NullableText)o2;
+ int rc = 0;
+
+ // If either are null, handle differently.
+ if (!nt1.isNull() && !nt2.isNull()) {
+ rc = ((String)nt1.getValueAsPigType()).compareTo((String)nt2.getValueAsPigType());
+ } else {
+ // For sorting purposes two nulls are equal.
+ if (nt1.isNull() && nt2.isNull()) rc = 0;
+ else if (nt1.isNull()) rc = -1;
+ else rc = 1;
+ }
+ if (!mAsc[0]) rc *= -1;
+ return rc;
+ }
+
+
}
Modified: incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigTupleRawComparator.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigTupleRawComparator.java?rev=698044&r1=698043&r2=698044&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigTupleRawComparator.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigTupleRawComparator.java Mon Sep 22 17:31:27 2008
@@ -33,7 +33,7 @@
import org.apache.pig.data.DataType;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;
-import org.apache.pig.impl.io.PigNullableWritable;
+import org.apache.pig.impl.io.NullableTuple;
import org.apache.pig.impl.util.ObjectSerializer;
public class PigTupleRawComparator extends WritableComparator implements Configurable {
@@ -100,36 +100,60 @@
throw new RuntimeException(ioe.getMessage(), ioe);
}
- int sz1 = t1.size();
- int sz2 = t2.size();
- if (sz2 < sz1) {
- rc = 1;
- } else if (sz2 > sz1) {
- rc = -1;
- } else {
- for (int i = 0; i < sz1; i++) {
- try {
- int c = DataType.compare(t1.get(i), t2.get(i));
- if (c != 0) {
- if (!mWholeTuple && !mAsc[i]) c *= -1;
- else if (mWholeTuple && !mAsc[0]) c *= -1;
- return c;
- }
- } catch (ExecException e) {
- throw new RuntimeException("Unable to compare tuples", e);
- }
- }
- rc = 0;
- }
+ rc = compareTuple(t1, t2);
+
} else {
// For sorting purposes two nulls are equal.
if (b1[s1] != 0 && b2[s2] != 0) rc = 0;
else if (b1[s1] != 0) rc = -1;
else rc = 1;
+ if (mWholeTuple && !mAsc[0]) rc *= -1;
+ }
+ return rc;
+ }
+
+ public int compare(Object o1, Object o2) {
+ NullableTuple nt1 = (NullableTuple)o1;
+ NullableTuple nt2 = (NullableTuple)o2;
+ int rc = 0;
+
+ // If either are null, handle differently.
+ if (!nt1.isNull() && !nt2.isNull()) {
+ rc = compareTuple((Tuple)nt1.getValueAsPigType(),
+ (Tuple)nt2.getValueAsPigType());
+ } else {
+ // For sorting purposes two nulls are equal.
+ if (nt1.isNull() && nt2.isNull()) rc = 0;
+ else if (nt1.isNull()) rc = -1;
+ else rc = 1;
+ if (mWholeTuple && !mAsc[0]) rc *= -1;
}
- if (mWholeTuple && !mAsc[0]) rc *= -1;
return rc;
}
+ private int compareTuple(Tuple t1, Tuple t2) {
+ int sz1 = t1.size();
+ int sz2 = t2.size();
+ if (sz2 < sz1) {
+ return 1;
+ } else if (sz2 > sz1) {
+ return -1;
+ } else {
+ for (int i = 0; i < sz1; i++) {
+ try {
+ int c = DataType.compare(t1.get(i), t2.get(i));
+ if (c != 0) {
+ if (!mWholeTuple && !mAsc[i]) c *= -1;
+ else if (mWholeTuple && !mAsc[0]) c *= -1;
+ return c;
+ }
+ } catch (ExecException e) {
+ throw new RuntimeException("Unable to compare tuples", e);
+ }
+ }
+ return 0;
+ }
+ }
+
}
Modified: incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/SortPartitioner.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/SortPartitioner.java?rev=698044&r1=698043&r2=698044&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/SortPartitioner.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/SortPartitioner.java Mon Sep 22 17:31:27 2008
@@ -53,6 +53,7 @@
int index = Arrays.binarySearch(quantiles, key, comparator);
if (index < 0)
index = -index-1;
+ // Shouldn't this be index % numPartitions?
return Math.min(index, numPartitions - 1);
}
@@ -84,7 +85,6 @@
}
}
convertToArray(job, quantiles);
- //this.quantiles = quantiles.toArray(new NullableTuple[0]);
}catch (Exception e){
throw new RuntimeException(e);
}
@@ -108,13 +108,6 @@
}
q.add(HDataType.getWritableComparableTypes(o,
Byte.valueOf(kts)));
- /*
- if (o == null) {
- q.add(new NullableTuple(t));
- } else {
- q.add(HDataType.getWritableComparableTypes(o, DataType.findType(o)));
- }
- */
}
} catch (Exception e) {
throw new RuntimeException(e);
@@ -142,11 +135,6 @@
} else {
throw new RuntimeException("Unexpected class in SortPartitioner");
}
-
-System.out.println("Quantiles:");
-for (int i = 0; i < quantiles.length; i++) {
-System.out.println(quantiles[i]);
-}
}
}
Modified: incubator/pig/branches/types/src/org/apache/pig/impl/builtin/FindQuantiles.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/impl/builtin/FindQuantiles.java?rev=698044&r1=698043&r2=698044&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/impl/builtin/FindQuantiles.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/impl/builtin/FindQuantiles.java Mon Sep 22 17:31:27 2008
@@ -25,20 +25,73 @@
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.data.BagFactory;
import org.apache.pig.data.DataBag;
+import org.apache.pig.data.DataType;
import org.apache.pig.data.Tuple;
public class FindQuantiles extends EvalFunc<DataBag>{
BagFactory mBagFactory = BagFactory.getInstance();
+ boolean[] mAsc;
+ enum State { ALL_ASC, ALL_DESC, MIXED };
+ State mState;
private class SortComparator implements Comparator<Tuple> {
public int compare(Tuple t1, Tuple t2) {
- return t1.compareTo(t2);
+ switch (mState) {
+ case ALL_ASC:
+ return t1.compareTo(t2);
+
+ case ALL_DESC:
+ return t2.compareTo(t1);
+
+ case MIXED:
+ // Have to break the tuple down and compare it field to field.
+ int sz1 = t1.size();
+ int sz2 = t2.size();
+ if (sz2 < sz1) {
+ return 1;
+ } else if (sz2 > sz1) {
+ return -1;
+ } else {
+ for (int i = 0; i < sz1; i++) {
+ try {
+ int c = DataType.compare(t1.get(i), t2.get(i));
+ if (c != 0) {
+ if (!mAsc[i]) c *= -1;
+ return c;
+ }
+ } catch (ExecException e) {
+ throw new RuntimeException("Unable to compare tuples", e);
+ }
+ }
+ return 0;
+ }
+ }
+ return -1; // keep the compiler happy
}
}
private Comparator<Tuple> mComparator = new SortComparator();
+ public FindQuantiles() {
+ mState = State.ALL_ASC;
+ }
+
+ public FindQuantiles(String[] args) {
+ mAsc = new boolean[args.length];
+ boolean sawAsc = false;
+ boolean sawDesc = false;
+ for (int i = 0; i < args.length; i++) {
+ mAsc[i] = Boolean.parseBoolean(args[i]);
+ if (mAsc[i]) sawAsc = true;
+ else sawDesc = true;
+ }
+ if (sawAsc && sawDesc) mState = State.MIXED;
+ else if (sawDesc) mState = State.ALL_DESC;
+ else mState = State.ALL_ASC; // In cast they gave us no args this
+ // defaults to all ascending.
+ }
+
/**
* first field in the input tuple is the number of quantiles to generate
* second field is the *sorted* bag of samples
Modified: incubator/pig/branches/types/test/org/apache/pig/test/TestEvalPipeline.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/test/org/apache/pig/test/TestEvalPipeline.java?rev=698044&r1=698043&r2=698044&view=diff
==============================================================================
--- incubator/pig/branches/types/test/org/apache/pig/test/TestEvalPipeline.java (original)
+++ incubator/pig/branches/types/test/org/apache/pig/test/TestEvalPipeline.java Mon Sep 22 17:31:27 2008
@@ -306,10 +306,12 @@
+ /*
@Test
public void testSort() throws Exception{
testSortDistinct(false, false);
}
+ */
@Test
public void testSortWithUDF() throws Exception{
@@ -366,7 +368,6 @@
assertFalse(seen.contains(act));
seen.add(act);
}else{
-System.out.println(last + " " + t.get(0));
assertTrue(last.compareTo(t.get(0).toString())<=0);
assertEquals(t.size(), 2);
last = t.get(0).toString();
Modified: incubator/pig/branches/types/test/org/apache/pig/test/TestMRCompiler.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/test/org/apache/pig/test/TestMRCompiler.java?rev=698044&r1=698043&r2=698044&view=diff
==============================================================================
--- incubator/pig/branches/types/test/org/apache/pig/test/TestMRCompiler.java (original)
+++ incubator/pig/branches/types/test/org/apache/pig/test/TestMRCompiler.java Mon Sep 22 17:31:27 2008
@@ -700,7 +700,7 @@
POUserComparisonFunc comparator = new POUserComparisonFunc(
new OperatorKey("", r.nextLong()), -1, null, new FuncSpec(funcName));
POSort sort = new POSort(new OperatorKey("", r.nextLong()), -1, ldFil1.getLeaves(),
- null, null, comparator);
+ null, new ArrayList<Boolean>(), comparator);
sort.setRequestedParallelism(20);
PhysicalPlan nesSortPlan = new PhysicalPlan();
POProject topPrj = new POProject(new OperatorKey("", r.nextLong()));