From a0b5457478f5795f3683a04c93b93358366381a4 Mon Sep 17 00:00:00 2001
From: Adam Cataldo <adamc@wealthfront.com>
Date: Thu, 1 Aug 2013 11:12:31 -0700
Subject: [PATCH] Adding EqualsValue filter

---
 .../operation/filter/EqualsValue.java         | 47 +++++++++++++++++++
 .../operation/filter/EqualsValueTest.java     | 38 +++++++++++++++
 2 files changed, 85 insertions(+)
 create mode 100644 src/core/cascading/operation/filter/EqualsValue.java
 create mode 100644 src/test/cascading/operation/filter/EqualsValueTest.java

diff --git a/src/core/cascading/operation/filter/EqualsValue.java b/src/core/cascading/operation/filter/EqualsValue.java
new file mode 100644
index 0000000000..75fbc49b67
--- /dev/null
+++ b/src/core/cascading/operation/filter/EqualsValue.java
@@ -0,0 +1,47 @@
+package cascading.operation.filter;
+
+import cascading.flow.FlowProcess;
+import cascading.operation.BaseOperation;
+import cascading.operation.Filter;
+import cascading.operation.FilterCall;
+import cascading.tuple.TupleEntry;
+
+/**
+ * Class EqualsValue verifies that the first n values in the argument values {@link cascading.tuple.Tuple} equals the n
+ * values specified by the constructor. If a non-equal value is encountered, or if the tuple does not have enough
+ * values, the current Tuple will be filtered out.
+ */
+public class EqualsValue extends BaseOperation<Void> implements Filter<Void> {
+
+  private static final long serialVersionUID = 1L;
+
+  private Object[] values;
+
+  public EqualsValue(Object value, Object... values) {
+    super(1);
+    this.values = new Object[values.length + 1];
+    this.values[0] = value;
+    System.arraycopy(values, 0, this.values, 1, values.length);
+  }
+
+  @Override
+  @SuppressWarnings("rawtypes")
+  public boolean isRemove(FlowProcess flowProcess, FilterCall<Void> filterCall) {
+    TupleEntry arguments = filterCall.getArguments();
+    if (arguments.size() < values.length) {
+      return true;
+    }
+    for (int i = 0; i < values.length; i++) {
+      Object arg = arguments.getObject(i);
+      if (values[i] == null || arg == null) {
+        if (values[i] != null || arg != null) {
+          return true;
+        }
+      }
+      if (!values[i].equals(arg)) {
+        return true;
+      }
+    }
+    return false;
+  }
+}
diff --git a/src/test/cascading/operation/filter/EqualsValueTest.java b/src/test/cascading/operation/filter/EqualsValueTest.java
new file mode 100644
index 0000000000..5fdda44ddb
--- /dev/null
+++ b/src/test/cascading/operation/filter/EqualsValueTest.java
@@ -0,0 +1,38 @@
+package cascading.operation.filter;
+
+import cascading.tuple.Tuple;
+import org.junit.Test;
+
+import static junit.framework.Assert.assertEquals;
+import static cascading.CascadingTestCase.invokeFilter;
+
+public class EqualsValueTest {
+  @Test
+  public void testEqualsValue() {
+    EqualsValue e1 = new EqualsValue("aaa");
+    Tuple[] args1 = new Tuple[] {
+        new Tuple("aaa"),
+        new Tuple("bbb"),
+        new Tuple(1),
+        new Tuple(null)
+    };
+    boolean[] actual1 = invokeFilter(e1, args1);
+    assertEquals(false, actual1[0]);
+    assertEquals(true, actual1[1]);
+    assertEquals(true, actual1[2]);
+    assertEquals(true, actual1[3]);
+    
+    EqualsValue e2 = new EqualsValue(1, "x");
+    Tuple[] args2 = new Tuple[] {
+        new Tuple(1, "x"),
+        new Tuple(2, "x"),
+        new Tuple(1, "y"),
+        new Tuple(1)
+    };
+    boolean[] actual2 = invokeFilter(e2, args2);
+    assertEquals(false, actual2[0]);
+    assertEquals(true, actual2[1]);
+    assertEquals(true, actual2[2]);
+    assertEquals(true, actual2[3]);
+  }
+}