Identify not range filters without negating subexpressions

sql/src/main/java/org/apache/druid/sql/calcite/filtration/CombineAndSimplifyBounds.java

@@ -41,6 +41,7 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 public class CombineAndSimplifyBounds extends BottomUpTransform
 {
@@ -63,25 +64,13 @@ public DimFilter process(DimFilter filter)
       return filter;
     } else if (filter instanceof AndDimFilter) {
       final List<DimFilter> children = getAndFilterChildren((AndDimFilter) filter);
-      final DimFilter one = doSimplifyAnd(children);
-      final DimFilter two = negate(doSimplifyOr(negateAll(children)));
-      return computeCost(one) <= computeCost(two) ? one : two;
+      return doSimplifyAnd(children);
     } else if (filter instanceof OrDimFilter) {
       final List<DimFilter> children = getOrFilterChildren((OrDimFilter) filter);
-      final DimFilter one = doSimplifyOr(children);
-      final DimFilter two = negate(doSimplifyAnd(negateAll(children)));
-      return computeCost(one) <= computeCost(two) ? one : two;
+      return doSimplifyOr(children);
     } else if (filter instanceof NotDimFilter) {
       final DimFilter field = ((NotDimFilter) filter).getField();
-      final DimFilter candidate;
-      if (field instanceof OrDimFilter) {
-        candidate = doSimplifyAnd(negateAll(getOrFilterChildren((OrDimFilter) field)));
-      } else if (field instanceof AndDimFilter) {
-        candidate = doSimplifyOr(negateAll(getAndFilterChildren((AndDimFilter) field)));
-      } else {
-        candidate = negate(field);
-      }
-      return computeCost(filter) <= computeCost(candidate) ? filter : candidate;
+      return negate(field);
     } else {
       return filter;
     }
@@ -166,6 +155,7 @@ private static DimFilter doSimplify(final List<DimFilter> children, boolean disj
               (c, existingType) -> ColumnType.leastRestrictiveType(existingType, range.getMatchValueType())
           );
         }
+
         final List<ObjectIntPair<RangeFilter>> filterList =
             ranges.computeIfAbsent(rangeRefKey, k -> new ArrayList<>());
         filterList.add(ObjectIntPair.of(range, childIndex));
@@ -210,6 +200,19 @@ private static DimFilter doSimplify(final List<DimFilter> children, boolean disj
             childrenToAdd.add(Bounds.toFilter(boundRefKey, range));
           }
         }
+      } else if (disjunction && Range.all().equals(rangeSet.span())) {
+        // ranges in disjunction - spanning ALL
+        // complementer must be a negated set of ranges
+        for (final ObjectIntPair<BoundDimFilter> boundAndChildIndex : filterList) {
+          childrenToRemove.add(boundAndChildIndex.rightInt());
+        }
+        Set<Range<BoundValue>> newRanges = rangeSet.complement().asRanges();
+        List<DimFilter> newFilters = new ArrayList<>();
+        for (Range<BoundValue> range : newRanges) {
+          BoundDimFilter filter = Bounds.toFilter(boundRefKey, range);
+          newFilters.add(filter);
+        }
+        childrenToAdd.add(new NotDimFilter(disjunction(newFilters)));
       }
     }
 
@@ -270,6 +273,19 @@ private static DimFilter doSimplify(final List<DimFilter> children, boolean disj
             childrenToAdd.add(Ranges.toFilter(rangeRefKey, range));
           }
         }
+      } else if (disjunction && Range.all().equals(rangeSet.span())) {
+        // ranges in disjunction - spanning ALL
+        // complementer must be a negated set of ranges
+        for (final ObjectIntPair<RangeFilter> boundAndChildIndex : filterList) {
+          childrenToRemove.add(boundAndChildIndex.rightInt());
+        }
+        Set<Range<RangeValue>> newRanges = rangeSet.complement().asRanges();
+        List<DimFilter> newFilters = new ArrayList<>();
+        for (Range<RangeValue> range : newRanges) {
+          RangeFilter filter = Ranges.toFilter(rangeRefKey, range);
+          newFilters.add(filter);
+        }
+        childrenToAdd.add(new NotDimFilter(disjunction(newFilters)));
       }
     }
 
@@ -327,6 +343,15 @@ private static DimFilter doSimplify(final List<DimFilter> children, boolean disj
     }
   }
 
+  private static DimFilter disjunction(List<DimFilter> operands)
+  {
+    Preconditions.checkArgument(operands.size() > 0, "invalid number of operands");
+    if (operands.size() == 1) {
+      return operands.get(0);
+    }
+    return new OrDimFilter(operands);
+  }
+
   private static DimFilter negate(final DimFilter filter)
   {
     if (Filtration.matchEverything().equals(filter)) {
@@ -345,34 +370,4 @@ private static DimFilter negate(final DimFilter filter)
       return new NotDimFilter(filter);
     }
   }
-
-  private static List<DimFilter> negateAll(final List<DimFilter> children)
-  {
-    final List<DimFilter> newChildren = Lists.newArrayListWithCapacity(children.size());
-    for (final DimFilter child : children) {
-      newChildren.add(negate(child));
-    }
-    return newChildren;
-  }
-
-  private static int computeCost(final DimFilter filter)
-  {
-    if (filter instanceof NotDimFilter) {
-      return computeCost(((NotDimFilter) filter).getField());
-    } else if (filter instanceof AndDimFilter) {
-      int cost = 0;
-      for (DimFilter field : ((AndDimFilter) filter).getFields()) {
-        cost += computeCost(field);
-      }
-      return cost;
-    } else if (filter instanceof OrDimFilter) {
-      int cost = 0;
-      for (DimFilter field : ((OrDimFilter) filter).getFields()) {
-        cost += computeCost(field);
-      }
-      return cost;
-    } else {
-      return 1;
-    }
-  }
 }

sql/src/test/java/org/apache/druid/sql/calcite/filtration/CombineAndSimplifyBoundsTest.java

@@ -0,0 +1,170 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.sql.calcite.filtration;
+
+import com.google.common.collect.ImmutableList;
+import org.apache.druid.query.filter.BoundDimFilter;
+import org.apache.druid.query.filter.DimFilter;
+import org.apache.druid.query.filter.RangeFilter;
+import org.apache.druid.segment.column.ColumnType;
+import org.apache.druid.sql.calcite.BaseCalciteQueryTest;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
+
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+
+@RunWith(Parameterized.class)
+public class CombineAndSimplifyBoundsTest extends BaseCalciteQueryTest
+{
+
+  enum RangeFilterType
+  {
+    BOUND {
+      @Override
+      public DimFilter range(String lit1, boolean gte, String name, boolean lte, String lit2)
+      {
+        return new BoundDimFilter(
+            name,
+            lit1,
+            lit2,
+            !gte,
+            !lte,
+            true,
+            null,
+            null
+        );
+      }
+    },
+    RANGE {
+      @Override
+      public DimFilter range(String lit1, boolean gte, String name, boolean lte, String lit2)
+      {
+        return new RangeFilter(
+            name,
+            ColumnType.STRING,
+            lit1,
+            lit2,
+            !gte,
+            !lte,
+            null
+        );
+      }
+    };
+
+    public final DimFilter lt(String name, String literal)
+    {
+      return range(null, false, name, false, literal);
+    }
+
+    public final DimFilter le(String name, String literal)
+    {
+      return range(null, false, name, true, literal);
+    }
+
+    public final DimFilter gt(String name, String literal)
+    {
+      return range(literal, false, name, false, null);
+    }
+
+    public final DimFilter ge(String name, String literal)
+    {
+      return range(literal, true, name, false, null);
+    }
+
+    public final DimFilter eq(String name, String literal)
+    {
+      return range(literal, true, name, true, literal);
+    }
+
+    public abstract DimFilter range(String lit1, boolean gte, String name, boolean lte, String lit2);
+  }
+
+  @Parameters
+  public static List<Object[]> getParameters()
+  {
+    return ImmutableList.of(
+        new Object[] {RangeFilterType.BOUND},
+        new Object[] {RangeFilterType.RANGE}
+    );
+  }
+
+  final RangeFilterType rangeFilter;
+
+  public CombineAndSimplifyBoundsTest(RangeFilterType rangeFilter)
+  {
+    this.rangeFilter = rangeFilter;
+  }
+
+  @Test
+  public void testNotAZ()
+  {
+    String dim1 = "dim1";
+    DimFilter inputFilter = or(
+        rangeFilter.lt(dim1, "a"),
+        rangeFilter.gt(dim1, "z")
+    );
+    DimFilter expected = not(rangeFilter.range("a", true, dim1, true, "z"));
+
+    check(inputFilter, expected);
+  }
+
+  @Test
+  public void testAZ()
+  {
+    String dim1 = "dim1";
+    DimFilter inputFilter = and(
+        rangeFilter.gt(dim1, "a"),
+        rangeFilter.lt(dim1, "z")
+    );
+    DimFilter expected = rangeFilter.range("a", false, dim1, false, "z");
+
+    check(inputFilter, expected);
+  }
+
+  @Test
+  public void testNot2()
+  {
+    String dim1 = "dim1";
+    DimFilter inputFilter = or(
+        rangeFilter.le(dim1, "a"),
+        rangeFilter.range("f", true, dim1, false, "j"),
+        rangeFilter.gt(dim1, "z")
+    );
+    DimFilter expected = not(
+        or(
+            rangeFilter.range("a", false, dim1, false, "f"),
+            rangeFilter.range("j", true, dim1, true, "z")
+        )
+    );
+
+    check(inputFilter, expected);
+  }
+
+  private void check(DimFilter inputFilter, DimFilter expected)
+  {
+    Filtration filt = Filtration.create(inputFilter, null);
+    Filtration ret = CombineAndSimplifyBounds.instance().apply(filt);
+    assertEquals(expected, ret.getDimFilter());
+  }
+}