[CALCITE-4446] Implement three-valued logic for SEARCH operator

Close apache#2357

core/src/main/java/org/apache/calcite/rel/rel2sql/SqlImplementor.java

@@ -47,6 +47,7 @@
 import org.apache.calcite.rex.RexProgram;
 import org.apache.calcite.rex.RexShuttle;
 import org.apache.calcite.rex.RexSubQuery;
+import org.apache.calcite.rex.RexUnknownAs;
 import org.apache.calcite.rex.RexWindow;
 import org.apache.calcite.rex.RexWindowBound;
 import org.apache.calcite.sql.JoinType;
@@ -892,7 +893,7 @@ private <C extends Comparable<C>> SqlNode toSql(@Nullable RexProgram program,
         RexNode operand, RelDataType type, Sarg<C> sarg) {
       final List<SqlNode> orList = new ArrayList<>();
       final SqlNode operandSql = toSql(program, operand);
-      if (sarg.containsNull) {
+      if (sarg.nullAs == RexUnknownAs.TRUE) {
         orList.add(SqlStdOperatorTable.IS_NULL.createCall(POS, operandSql));
       }
       if (sarg.isPoints()) {

core/src/main/java/org/apache/calcite/rex/RexAnalyzer.java

@@ -131,11 +131,14 @@ private static class VariableCollector extends RexVisitorImpl<Void> {
     }
 
     @Override public Void visitCall(RexCall call) {
-      if (!RexInterpreter.SUPPORTED_SQL_KIND.contains(call.getKind())) {
+      switch (call.getKind()) {
+      case CAST:
+      case OTHER_FUNCTION:
         ++unsupportedCount;
         return null;
+      default:
+        return super.visitCall(call);
       }
-      return super.visitCall(call);
     }
   }
 }

core/src/main/java/org/apache/calcite/rex/RexBuilder.java

@@ -1330,7 +1330,7 @@ public RexNode makeNullLiteral(SqlTypeName typeName) {
    * otherwise creates a disjunction, "arg = point0 OR arg = point1 OR ...". */
   public RexNode makeIn(RexNode arg, List<? extends RexNode> ranges) {
     if (areAssignable(arg, ranges)) {
-      final Sarg sarg = toSarg(Comparable.class, ranges, false);
+      final Sarg sarg = toSarg(Comparable.class, ranges, RexUnknownAs.UNKNOWN);
       if (sarg != null) {
         final RexNode range0 = ranges.get(0);
         return makeCall(SqlStdOperatorTable.SEARCH,
@@ -1369,7 +1369,7 @@ public RexNode makeBetween(RexNode arg, RexNode lower, RexNode upper) {
         && upperValue != null
         && areAssignable(arg, Arrays.asList(lower, upper))) {
       final Sarg sarg =
-          Sarg.of(false,
+          Sarg.of(RexUnknownAs.UNKNOWN,
               ImmutableRangeSet.<Comparable>of(
                   Range.closed(lowerValue, upperValue)));
       return makeCall(SqlStdOperatorTable.SEARCH, arg,
@@ -1384,7 +1384,7 @@ && areAssignable(arg, Arrays.asList(lower, upper))) {
    * not possible. */
   @SuppressWarnings({"BetaApi", "UnstableApiUsage"})
   private static <C extends Comparable<C>> @Nullable Sarg<C> toSarg(Class<C> clazz,
-      List<? extends RexNode> ranges, boolean containsNull) {
+      List<? extends RexNode> ranges, RexUnknownAs unknownAs) {
     if (ranges.isEmpty()) {
       // Cannot convert an empty list to a Sarg (by this interface, at least)
       // because we use the type of the first element.
@@ -1398,7 +1398,7 @@ && areAssignable(arg, Arrays.asList(lower, upper))) {
       }
       rangeSet.add(Range.singleton(value));
     }
-    return Sarg.of(containsNull, rangeSet);
+    return Sarg.of(unknownAs, rangeSet);
   }
 
   private static <C extends Comparable<C>> @Nullable C toComparable(Class<C> clazz,

core/src/main/java/org/apache/calcite/rex/RexCall.java

@@ -214,7 +214,8 @@ private boolean digestWithType() {
     case SEARCH:
       final Sarg sarg = ((RexLiteral) operands.get(1)).getValueAs(Sarg.class);
       return requireNonNull(sarg, "sarg").isAll()
-          && (sarg.containsNull || !operands.get(0).getType().isNullable());
+          && (sarg.nullAs == RexUnknownAs.TRUE
+              || !operands.get(0).getType().isNullable());
     default:
       return false;
     }
@@ -235,7 +236,8 @@ private boolean digestWithType() {
     case SEARCH:
       final Sarg sarg = ((RexLiteral) operands.get(1)).getValueAs(Sarg.class);
       return requireNonNull(sarg, "sarg").isNone()
-          && (!sarg.containsNull || !operands.get(0).getType().isNullable());
+          && (sarg.nullAs == RexUnknownAs.FALSE
+              || !operands.get(0).getType().isNullable());
     default:
       return false;
     }

core/src/main/java/org/apache/calcite/rex/RexInterpreter.java

@@ -20,11 +20,19 @@
 import org.apache.calcite.avatica.util.TimeUnit;
 import org.apache.calcite.avatica.util.TimeUnitRange;
 import org.apache.calcite.rel.metadata.NullSentinel;
+import org.apache.calcite.runtime.SqlFunctions;
 import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.calcite.util.DateString;
 import org.apache.calcite.util.NlsString;
+import org.apache.calcite.util.RangeSets;
+import org.apache.calcite.util.Sarg;
+import org.apache.calcite.util.TimeString;
+import org.apache.calcite.util.TimestampString;
 import org.apache.calcite.util.Util;
 
 import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.RangeSet;
 
 import org.checkerframework.checker.nullness.qual.Nullable;
 
@@ -209,6 +217,10 @@ private Comparable getOrUnbound(RexNode e) {
       return ceil(call, values);
     case EXTRACT:
       return extract(values);
+    case LIKE:
+      return like(values);
+    case SEARCH:
+      return search(call.operands.get(1).getType().getSqlTypeName(), values);
     default:
       throw unbound(call);
     }
@@ -231,6 +243,58 @@ private static Comparable extract(List<Comparable> values) {
     return DateTimeUtils.unixDateExtract(timeUnitRange, v2);
   }
 
+  private static Comparable like(List<Comparable> values) {
+    if (containsNull(values)) {
+      return N;
+    }
+    final NlsString value = (NlsString) values.get(0);
+    final NlsString pattern = (NlsString) values.get(1);
+    switch (values.size()) {
+    case 2:
+      return SqlFunctions.like(value.getValue(), pattern.getValue());
+    case 3:
+      final NlsString escape = (NlsString) values.get(2);
+      return SqlFunctions.like(value.getValue(), pattern.getValue(),
+          escape.getValue());
+    default:
+      throw new AssertionError();
+    }
+  }
+
+  @SuppressWarnings({"BetaApi", "rawtypes", "unchecked", "UnstableApiUsage"})
+  private static Comparable search(SqlTypeName typeName, List<Comparable> values) {
+    final Comparable value = values.get(0);
+    final Sarg sarg = (Sarg) values.get(1);
+    if (value == N) {
+      switch (sarg.nullAs) {
+      case FALSE:
+        return false;
+      case TRUE:
+        return true;
+      default:
+        return N;
+      }
+    }
+    return translate(sarg.rangeSet, typeName).contains(value);
+  }
+
+  /** Translates the values in a RangeSet from literal format to runtime format.
+   * For example the DATE SQL type uses DateString for literals and Integer at
+   * runtime. */
+  @SuppressWarnings({"BetaApi", "rawtypes", "unchecked", "UnstableApiUsage"})
+  private static RangeSet translate(RangeSet rangeSet, SqlTypeName typeName) {
+    switch (typeName) {
+    case DATE:
+      return RangeSets.copy(rangeSet, DateString::getDaysSinceEpoch);
+    case TIME:
+      return RangeSets.copy(rangeSet, TimeString::getMillisOfDay);
+    case TIMESTAMP:
+      return RangeSets.copy(rangeSet, TimestampString::getMillisSinceEpoch);
+    default:
+      return rangeSet;
+    }
+  }
+
   private static Comparable coalesce(List<Comparable> values) {
     for (Comparable value : values) {
       if (value != N) {