package humboldt;

import java.util.Arrays;

import org.apache.hadoop.conf.Configurable;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.BinaryComparable;
import org.apache.hadoop.io.WritableUtils;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.mapreduce.Job;


public class BinaryComparator<K, V> extends WritableComparator implements Configurable {
  public static String LEFT_OFFSET_PROPERTY_NAME = "humboldt.binarycomparator.left.offset";
  public static String RIGHT_OFFSET_PROPERTY_NAME = "humboldt.binarycomparator.right.offset";

  private Configuration conf;
  private int leftOffset;
  private int rightOffset;

  public BinaryComparator() {
    super(BytesWritable.class);
  }

  public static void setOffsets(Configuration conf, int left, int right) {
    conf.setInt(LEFT_OFFSET_PROPERTY_NAME, left);
    conf.setInt(RIGHT_OFFSET_PROPERTY_NAME, right);
  }

  public static void setLeftOffset(Configuration conf, int offset) {
    conf.setInt(LEFT_OFFSET_PROPERTY_NAME, offset);
  }

  public static void setRightOffset(Configuration conf, int offset) {
    conf.setInt(RIGHT_OFFSET_PROPERTY_NAME, offset);
  }

  public void setConf(Configuration conf) {
    this.conf = conf;
    leftOffset = conf.getInt(LEFT_OFFSET_PROPERTY_NAME, 0);
    rightOffset = conf.getInt(RIGHT_OFFSET_PROPERTY_NAME, -1);
  }

  public Configuration getConf() {
    return conf;
  }

  public int compare(WritableComparable w1, WritableComparable w2) {
    BinaryComparable bc1 = (BinaryComparable) w1;
    BinaryComparable bc2 = (BinaryComparable) w2;
    return compare(bc1.getBytes(), 0, bc1.getLength(),
                   bc2.getBytes(), 0, bc2.getLength());
  }

  public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
    int b1LeftIndex = (leftOffset + l1) % l1;
    int b1RightIndex = (rightOffset + l1) % l1;
    int b2LeftIndex = (leftOffset + l2) % l2;
    int b2RightIndex = (rightOffset + l2) % l2;
    return WritableComparator.compareBytes(b1, b1LeftIndex, b1RightIndex - b1LeftIndex + 1,
                                           b2, b2LeftIndex, b2RightIndex - b2LeftIndex + 1);
  }
}

@@ -16,34 +16,8 @@
 t;%0A%0A
-import java.util.Arrays;%0A%0A
 impo
@@ -325,24 +325,58 @@
 esWritable;%0A
+import org.apache.hadoop.io.Text;%0A
 import org.a
@@ -403,16 +403,64 @@
 ce.Job;%0A
+import org.apache.hadoop.mapreduce.MRJobConfig;%0A
 %0A%0Apublic
@@ -759,16 +759,70 @@
 n conf;%0A
+  private Class%3C? extends BinaryComparable%3E keyClass;%0A
   privat
@@ -911,23 +911,108 @@
 per(
-BytesWritable.c
+null);%0A  %7D%0A%0A  public BinaryComparator(Class%3C? extends WritableComparable%3E keyClass) %7B%0A    super(keyC
 lass
@@ -1525,16 +1525,21 @@
 nf;%0A    
+this.
 leftOffs
@@ -1590,16 +1590,21 @@
 0);%0A    
+this.
 rightOff
@@ -1650,24 +1650,140 @@
 _NAME, -1);%0A
+    this.keyClass = conf.getClass(MRJobConfig.MAP_OUTPUT_VALUE_CLASS, BytesWritable.class, BinaryComparable.class);%0A
   %7D%0A%0A  publi
@@ -2019,16 +2019,24 @@
  compare
+RawBytes
 (bc1.get
@@ -2064,16 +2064,24 @@
 ngth(),%0A
+        
         
@@ -2206,24 +2206,461 @@
 , int l2) %7B%0A
+    int n1 = 0;%0A    int n2 = 0;%0A    if (keyClass == BytesWritable.class) %7B%0A      n1 = 4;%0A      n2 = 4;%0A    %7D else if (keyClass == Text.class) %7B%0A      n1 = WritableUtils.decodeVIntSize(b1%5Bs1%5D);%0A      n2 = WritableUtils.decodeVIntSize(b2%5Bs2%5D);%0A    %7D%0A    return compareRawBytes(b1, s1 + n1, l1 - n1,%0A                           b2, s2 + n2, l2 - n2);%0A  %7D%0A%0A  public int compareRawBytes(byte%5B%5D b1, int s1, int l1, byte%5B%5D b2, int s2, int l2) %7B%0A
     int b1Le
@@ -2674,32 +2674,37 @@
  = (leftOffset +
+ s1 +
  l1) %25 l1;%0A    i
@@ -2735,16 +2735,21 @@
 Offset +
+ s1 +
  l1) %25 l
@@ -2778,32 +2778,37 @@
  = (leftOffset +
+ s2 +
  l2) %25 l2;%0A    i
@@ -2839,16 +2839,21 @@
 Offset +
+ s2 +
  l2) %25 l

Support BinaryWritable and Text and auto detect which is used