Druid OR filters should not discard matching timestamp ranges

druid/src/main/java/org/apache/calcite/adapter/druid/DruidDateTimeUtils.java

@@ -126,9 +126,11 @@ protected static List<Interval> toInterval(
       for (RexNode child : call.getOperands()) {
         List<Range<Long>> extracted =
             extractRanges(child, withNot);
-        if (extracted != null) {
-          intervals.addAll(extracted);
+        if (extracted == null) {
+          // Every disjunct must be represented to avoid excluding matching rows.
+          return null;
         }
+        intervals.addAll(extracted);
       }
       return intervals;
     }

druid/src/test/java/org/apache/calcite/adapter/druid/DruidQueryFilterTest.java

@@ -16,14 +16,29 @@
  */
 package org.apache.calcite.adapter.druid;
 
+import org.apache.calcite.avatica.util.TimeUnitRange;
 import org.apache.calcite.config.CalciteConnectionConfig;
+import org.apache.calcite.config.CalciteConnectionConfigImpl;
+import org.apache.calcite.interpreter.BindableConvention;
 import org.apache.calcite.jdbc.JavaTypeFactoryImpl;
+import org.apache.calcite.plan.Contexts;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelOptTable;
+import org.apache.calcite.plan.hep.HepPlanner;
+import org.apache.calcite.plan.hep.HepProgramBuilder;
+import org.apache.calcite.prepare.RelOptTableImpl;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Filter;
+import org.apache.calcite.rel.logical.LogicalFilter;
+import org.apache.calcite.rel.logical.LogicalTableScan;
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rel.type.RelDataTypeSystem;
 import org.apache.calcite.rex.RexBuilder;
 import org.apache.calcite.rex.RexNode;
 import org.apache.calcite.sql.fun.SqlInternalOperators;
+import org.apache.calcite.sql.fun.SqlStdOperatorTable;
 import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.calcite.util.TimestampString;
 
 import com.fasterxml.jackson.core.JsonFactory;
 import com.fasterxml.jackson.core.JsonGenerator;
@@ -38,10 +53,13 @@
 import java.io.StringWriter;
 import java.math.BigDecimal;
 import java.util.List;
+import java.util.Properties;
 
+import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.CoreMatchers.notNullValue;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.hasToString;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
 
 /**
  * Tests generating Druid filters.
@@ -138,6 +156,42 @@ class DruidQueryFilterTest {
             + "\"ordering\":\"lexicographic\"}"));
   }
 
+  @Test void testOrWithExtractRetainsFilter() {
+    final Fixture f = new Fixture();
+    final HepPlanner planner =
+        new HepPlanner(new HepProgramBuilder().addRuleInstance(DruidRules.FILTER).build(),
+            Contexts.of(new CalciteConnectionConfigImpl(new Properties())));
+    final RelOptCluster cluster = RelOptCluster.create(planner, f.rexBuilder);
+    final RelDataType rowType = f.typeFactory.builder()
+        .add("timestamp", SqlTypeName.TIMESTAMP)
+        .build();
+    final DruidTable table =
+        new DruidTable(Mockito.mock(DruidSchema.class), "events", factory -> rowType,
+            ImmutableSet.of(), "timestamp", null, null, null);
+    final RelOptTable relOptTable =
+        RelOptTableImpl.create(null, rowType, ImmutableList.of("events"), table,
+            clazz -> null);
+    final RelNode scan = LogicalTableScan.create(cluster, relOptTable, ImmutableList.of());
+    final DruidQuery query =
+        DruidQuery.create(cluster, cluster.traitSet().plus(BindableConvention.INSTANCE),
+            relOptTable, table, ImmutableList.of(scan));
+    final RexNode timestamp = f.rexBuilder.makeInputRef(scan, 0);
+    final RexNode condition =
+        f.rexBuilder.makeCall(
+            SqlStdOperatorTable.OR, f.rexBuilder.makeCall(SqlStdOperatorTable.LESS_THAN, timestamp,
+            f.rexBuilder.makeTimestampLiteral(new TimestampString("2020-01-01 00:00:00"), 0)),
+        f.rexBuilder.makeCall(SqlStdOperatorTable.EQUALS,
+            f.rexBuilder.makeCall(SqlStdOperatorTable.EXTRACT,
+                f.rexBuilder.makeFlag(TimeUnitRange.DAY), timestamp),
+            f.rexBuilder.makeExactLiteral(BigDecimal.valueOf(15))));
+    planner.setRoot(LogicalFilter.create(query, condition));
+
+    final DruidQuery result = assertInstanceOf(DruidQuery.class, planner.findBestExp());
+    assertThat(result.intervals, is(query.intervals));
+    final Filter filter = assertInstanceOf(Filter.class, result.getTopNode());
+    assertThat(filter.getCondition(), is(condition));
+  }
+
   /** Everything a test needs for a healthy, active life. */
   static class Fixture {
     final JavaTypeFactoryImpl typeFactory =

druid/src/test/java/org/apache/calcite/test/DruidDateRangeRulesTest.java

@@ -36,6 +36,7 @@
 
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.CoreMatchers.notNullValue;
+import static org.hamcrest.CoreMatchers.nullValue;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.hasToString;
 
@@ -59,6 +60,23 @@ class DruidDateRangeRulesTest {
         is("[2011-01-01T00:00:00.000Z/2012-02-02T00:00:00.001Z]"));
   }
 
+  @Test void testOrWithUnextractableRange() {
+    final Fixture2 f = new Fixture2();
+    final RexNode range = f.lt(f.ts, f.timestampLiteral(2020, Calendar.JANUARY, 1));
+    final RexNode extract = f.eq(f.exDay, f.literal(15));
+    assertThat(DruidDateTimeUtils.createInterval(f.or(range, extract)), nullValue());
+    assertThat(DruidDateTimeUtils.createInterval(f.or(extract, range)), nullValue());
+  }
+
+  @Test void testOrWithEmptyRange() {
+    final Fixture2 f = new Fixture2();
+    final RexNode timestamp = f.timestampLiteral(2020, Calendar.JANUARY, 1);
+    final RexNode empty = f.and(f.lt(f.ts, timestamp), f.ge(f.ts, timestamp));
+    checkDateRangeNoSimplify(f,
+        f.or(empty, f.eq(f.ts, timestamp)),
+        is("[2020-01-01T00:00:00.000Z/2020-01-01T00:00:00.001Z]"));
+  }
+
   @Test void testExtractYearAndDayFromDateColumn() {
     final Fixture2 f = new Fixture2();
     // AND(AND(>=($8, 2010-01-01), <($8, 2011-01-01)),