OrFilter: Properly handle child matchers that return the original mask. (#10754)

* OrFilter: Properly handle child matchers that return the original mask.

This happens when a child matcher is literally true (for example,
BooleanVectorValueMatcher). In this case, OrFilter would throw this
exception from its call to removeAll while processing the next filter:

  java.lang.IllegalStateException: 'other' must be a different instance from 'this'

Also update the javadocs for VectorValueMatcher to call out that the
returned object may be the same as the input mask.

* Fix style.

processing/src/main/java/org/apache/druid/query/filter/vector/VectorMatch.java

@@ -140,6 +140,8 @@ public class VectorMatch implements ReadableVectorMatch
   /**
    * Removes all rows from this object that occur in "other", in place, and returns a reference to this object. Does
    * not modify "other".
+   *
+   * "other" cannot be the same instance as this object.
    */
   public VectorMatch removeAll(final ReadableVectorMatch other)
   {

processing/src/main/java/org/apache/druid/query/filter/vector/VectorValueMatcher.java

@@ -34,9 +34,12 @@ public interface VectorValueMatcher extends VectorSizeInspector
   /**
    * Examine the current vector and return a match indicating what is accepted.
    *
+   * Does not modify "mask".
+   *
    * @param mask must not be null; use {@link VectorMatch#allTrue} if you don't need a mask.
    *
-   * @return the subset of "mask" that this value matcher accepts
+   * @return the subset of "mask" that this value matcher accepts. May be the same instance as {@param mask} if
+   * every row in the mask matches the filter.
    */
   ReadableVectorMatch match(ReadableVectorMatch mask);
 }

processing/src/main/java/org/apache/druid/segment/filter/OrFilter.java

@@ -212,7 +212,13 @@ public class OrFilter implements BooleanFilter
       {
         ReadableVectorMatch currentMatch = baseMatchers[0].match(mask);
 
+        // Initialize currentMask = mask, then progressively remove rows from the mask as we find matches for them.
+        // This isn't necessary for correctness (we could use the original "mask" on every call to "match") but it
+        // allows for short-circuiting on a row-by-row basis.
         currentMask.copyFrom(mask);
+
+        // Initialize retVal = currentMatch, the rows matched by the first matcher. We'll add more as we loop over
+        // the rest of the matchers.
         retVal.copyFrom(currentMatch);
 
         for (int i = 1; i < baseMatchers.length; i++) {
@@ -224,6 +230,11 @@ public class OrFilter implements BooleanFilter
           currentMask.removeAll(currentMatch);
           currentMatch = baseMatchers[i].match(currentMask);
           retVal.addAll(currentMatch, scratch);
+
+          if (currentMatch == currentMask) {
+            // baseMatchers[i] matched every remaining row. Short-circuit out.
+            break;
+          }
         }
 
         assert retVal.isValid(mask);

processing/src/test/java/org/apache/druid/segment/filter/OrFilterTest.java

@@ -19,14 +19,247 @@
 
 package org.apache.druid.segment.filter;
 
+import com.google.common.base.Function;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
 import nl.jqno.equalsverifier.EqualsVerifier;
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.impl.DimensionsSpec;
+import org.apache.druid.data.input.impl.InputRowParser;
+import org.apache.druid.data.input.impl.MapInputRowParser;
+import org.apache.druid.data.input.impl.TimeAndDimsParseSpec;
+import org.apache.druid.data.input.impl.TimestampSpec;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.query.filter.AndDimFilter;
+import org.apache.druid.query.filter.InDimFilter;
+import org.apache.druid.query.filter.NotDimFilter;
+import org.apache.druid.query.filter.OrDimFilter;
+import org.apache.druid.query.filter.SelectorDimFilter;
+import org.apache.druid.query.filter.TrueDimFilter;
+import org.apache.druid.segment.IndexBuilder;
+import org.apache.druid.segment.StorageAdapter;
+import org.junit.AfterClass;
 import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
 
-public class OrFilterTest
+import java.io.Closeable;
+import java.util.List;
+import java.util.Map;
+
+@RunWith(Parameterized.class)
+public class OrFilterTest extends BaseFilterTest
 {
+  private static final String TIMESTAMP_COLUMN = "timestamp";
+
+  private static final InputRowParser<Map<String, Object>> PARSER = new MapInputRowParser(
+      new TimeAndDimsParseSpec(
+          new TimestampSpec(TIMESTAMP_COLUMN, "iso", DateTimes.of("2000")),
+          new DimensionsSpec(null, null, null)
+      )
+  );
+
+  private static final List<InputRow> ROWS = ImmutableList.of(
+      PARSER.parseBatch(ImmutableMap.of("dim0", "0", "dim1", "0")).get(0),
+      PARSER.parseBatch(ImmutableMap.of("dim0", "1", "dim1", "0")).get(0),
+      PARSER.parseBatch(ImmutableMap.of("dim0", "2", "dim1", "0")).get(0),
+      PARSER.parseBatch(ImmutableMap.of("dim0", "3", "dim1", "0")).get(0),
+      PARSER.parseBatch(ImmutableMap.of("dim0", "4", "dim1", "0")).get(0),
+      PARSER.parseBatch(ImmutableMap.of("dim0", "5", "dim1", "0")).get(0)
+  );
+
+  public OrFilterTest(
+      String testName,
+      IndexBuilder indexBuilder,
+      Function<IndexBuilder, Pair<StorageAdapter, Closeable>> finisher,
+      boolean cnf,
+      boolean optimize
+  )
+  {
+    super(testName, ROWS, indexBuilder, finisher, cnf, optimize);
+  }
+
+  @AfterClass
+  public static void tearDown() throws Exception
+  {
+    BaseFilterTest.tearDown(AndFilterTest.class.getName());
+  }
+
+  @Test
+  public void testOneFilterMatchSome()
+  {
+    assertFilterMatches(
+        new OrDimFilter(
+            ImmutableList.of(
+                new SelectorDimFilter("dim0", "1", null)
+            )
+        ),
+        ImmutableList.of("1")
+    );
+  }
+
+  @Test
+  public void testOneFilterMatchAll()
+  {
+    assertFilterMatches(
+        new OrDimFilter(
+            ImmutableList.of(
+                new SelectorDimFilter("dim1", "0", null)
+            )
+        ),
+        ImmutableList.of("0", "1", "2", "3", "4", "5")
+    );
+  }
+
+  @Test
+  public void testOneFilterMatchNone()
+  {
+    assertFilterMatches(
+        new OrDimFilter(
+            ImmutableList.of(
+                new SelectorDimFilter("dim1", "7", null)
+            )
+        ),
+        ImmutableList.of()
+    );
+  }
+
+  @Test
+  public void testTwoFilterFirstMatchesAllSecondMatchesNone()
+  {
+    assertFilterMatches(
+        new OrDimFilter(
+            ImmutableList.of(
+                new SelectorDimFilter("dim1", "0", null),
+                new SelectorDimFilter("dim0", "7", null)
+            )
+        ),
+        ImmutableList.of("0", "1", "2", "3", "4", "5")
+    );
+  }
+
+  @Test
+  public void testTwoFilterFirstMatchesNoneSecondMatchesAll()
+  {
+    assertFilterMatches(
+        new OrDimFilter(
+            ImmutableList.of(
+                new SelectorDimFilter("dim0", "7", null),
+                new SelectorDimFilter("dim1", "0", null)
+            )
+        ),
+        ImmutableList.of("0", "1", "2", "3", "4", "5")
+    );
+  }
+
+  @Test
+  public void testTwoFilterFirstMatchesNoneSecondLiterallyTrue()
+  {
+    assertFilterMatches(
+        new OrDimFilter(
+            ImmutableList.of(
+                new SelectorDimFilter("dim0", "7", null),
+                TrueDimFilter.instance()
+            )
+        ),
+        ImmutableList.of("0", "1", "2", "3", "4", "5")
+    );
+  }
+
+  @Test
+  public void testTwoFilterFirstMatchesAllSecondMatchesAll()
+  {
+    assertFilterMatches(
+        new OrDimFilter(
+            ImmutableList.of(
+                new SelectorDimFilter("dim1", "0", null),
+                new NotDimFilter(new SelectorDimFilter("dim0", "7", null))
+            )
+        ),
+        ImmutableList.of("0", "1", "2", "3", "4", "5")
+    );
+  }
+
+  @Test
+  public void testTwoFilterFirstLiterallyTrueSecondMatchesAll()
+  {
+    assertFilterMatches(
+        new OrDimFilter(
+            ImmutableList.of(
+                TrueDimFilter.instance(),
+                new NotDimFilter(new SelectorDimFilter("dim0", "7", null))
+            )
+        ),
+        ImmutableList.of("0", "1", "2", "3", "4", "5")
+    );
+  }
+
+  @Test
+  public void testTwoFilterFirstMatchesSomeSecondMatchesNone()
+  {
+    assertFilterMatches(
+        new OrDimFilter(
+            ImmutableList.of(
+                new SelectorDimFilter("dim0", "3", null),
+                new SelectorDimFilter("dim1", "7", null)
+            )
+        ),
+        ImmutableList.of("3")
+    );
+  }
+
+  @Test
+  public void testTwoFilterFirstMatchesNoneSecondMatchesSome()
+  {
+    assertFilterMatches(
+        new OrDimFilter(
+            ImmutableList.of(
+                new SelectorDimFilter("dim1", "7", null),
+                new SelectorDimFilter("dim0", "3", null)
+            )
+        ),
+        ImmutableList.of("3")
+    );
+  }
+
+  @Test
+  public void testTwoFilterFirstMatchesNoneSecondMatchesNone()
+  {
+    assertFilterMatches(
+        new OrDimFilter(
+            ImmutableList.of(
+                new SelectorDimFilter("dim1", "7", null),
+                new SelectorDimFilter("dim0", "7", null)
+            )
+        ),
+        ImmutableList.of()
+    );
+  }
+
+  @Test
+  public void testThreeFilterFirstMatchesSomeSecondLiterallyTrueThirdMatchesNone()
+  {
+    assertFilterMatches(
+        new AndDimFilter(
+            new InDimFilter("dim0", ImmutableSet.of("0", "1", "2", "4", "5")),
+            new OrDimFilter(
+                ImmutableList.of(
+                    new SelectorDimFilter("dim0", "4", null),
+                    TrueDimFilter.instance(),
+                    new SelectorDimFilter("dim0", "7", null)
+                )
+            )
+        ),
+        ImmutableList.of("0", "1", "2", "4", "5")
+    );
+  }
+
   @Test
-  public void test_equals()
+  public void testEquals()
   {
+    EqualsVerifier.forClass(OrDimFilter.class).usingGetClass().withIgnoredFields("cachedOptimizedFilter").verify();
     EqualsVerifier.forClass(OrFilter.class).usingGetClass().withNonnullFields("filters").verify();
   }
 }