Fix filtering on boolean values in transformation (#9812)

* Fix filter on boolean value in Transform

* assert

* more descriptive test

* remove assert

* add assert for cached string; disable tests

* typo

core/src/main/java/org/apache/druid/java/util/common/granularity/PeriodGranularity.java

@@ -60,7 +60,7 @@ public class PeriodGranularity extends Granularity implements JsonSerializable
   )
   {
     this.period = Preconditions.checkNotNull(period, "period can't be null!");
-    Preconditions.checkArgument(!Period.ZERO.equals(period), "zero period is not acceptable in QueryGranularity!");
+    Preconditions.checkArgument(!Period.ZERO.equals(period), "zero period is not acceptable in PeriodGranularity!");
     this.chronology = tz == null ? ISOChronology.getInstanceUTC() : ISOChronology.getInstance(tz);
     if (origin == null) {
       // default to origin in given time zone when aligning multi-period granularities

core/src/main/java/org/apache/druid/math/expr/Expr.java

@@ -1287,7 +1287,6 @@ abstract class BinaryEvalOpExprBase extends BinaryOpExprBase
       return ExprEval.of(null);
     }
 
-
     if (leftVal.type() == ExprType.STRING && rightVal.type() == ExprType.STRING) {
       return evalString(leftVal.asString(), rightVal.asString());
     } else if (leftVal.type() == ExprType.LONG && rightVal.type() == ExprType.LONG) {

core/src/main/java/org/apache/druid/math/expr/ExprEval.java

@@ -117,7 +117,7 @@ public abstract class ExprEval<T>
   }
 
   // Cached String values
-  private boolean stringValueValid = false;
+  private boolean stringValueCached = false;
   @Nullable
   private String stringValue;
 
@@ -137,17 +137,35 @@ public abstract class ExprEval<T>
     return value;
   }
 
+  void cacheStringValue(@Nullable String value)
+  {
+    stringValue = value;
+    stringValueCached = true;
+  }
+
+  @Nullable
+  String getCachedStringValue()
+  {
+    assert stringValueCached;
+    return stringValue;
+  }
+
+  boolean isStringValueCached()
+  {
+    return stringValueCached;
+  }
+
   @Nullable
   public String asString()
   {
-    if (!stringValueValid) {
+    if (!stringValueCached) {
       if (value == null) {
         stringValue = null;
       } else {
         stringValue = String.valueOf(value);
       }
 
-      stringValueValid = true;
+      stringValueCached = true;
     }
 
     return stringValue;
@@ -567,6 +585,21 @@ public abstract class ExprEval<T>
       super(value);
     }
 
+    @Override
+    @Nullable
+    public String asString()
+    {
+      if (!isStringValueCached()) {
+        if (value == null) {
+          cacheStringValue(null);
+        } else {
+          cacheStringValue(Arrays.toString(value));
+        }
+      }
+
+      return getCachedStringValue();
+    }
+
     @Override
     public boolean isNumericNull()
     {

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

@@ -116,11 +116,13 @@ public class PredicateValueMatcherFactory implements ColumnProcessorFactory<Valu
           } else if (rowValue instanceof Number) {
             // Double or some other non-int, non-long, non-float number.
             return getDoublePredicate().applyDouble((double) rowValue);
-          } else if (rowValue instanceof String || rowValue instanceof List) {
+          } else {
-            // String or list-of-something. Cast to list of strings and evaluate them as strings.
+            // Other types. Cast to list of strings and evaluate them as strings.
+            // Boolean values are handled here as well since it is not a known type in Druid.
             final List<String> rowValueStrings = Rows.objectToStrings(rowValue);
 
             if (rowValueStrings.isEmpty()) {
+              // Empty list is equivalent to null.
               return getStringPredicate().apply(null);
             }
 
@@ -131,9 +133,6 @@ public class PredicateValueMatcherFactory implements ColumnProcessorFactory<Valu
             }
 
             return false;
-          } else {
-            // Unfilterable type. Treat as null.
-            return getStringPredicate().apply(null);
           }
         }
 

processing/src/main/java/org/apache/druid/segment/virtual/ExpressionSelectors.java

@@ -265,7 +265,6 @@ public class ExpressionSelectors
           @Override
           protected String getValue()
           {
-
             return NullHandling.emptyToNullIfNeeded(baseSelector.getObject().asString());
           }
 

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

@@ -0,0 +1,429 @@
+/*
+ * 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.segment.filter;
+
+import com.google.common.collect.ImmutableList;
+import org.apache.druid.common.config.NullHandling;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.query.filter.SelectorPredicateFactory;
+import org.apache.druid.query.filter.ValueMatcher;
+import org.apache.druid.segment.DimensionSelector;
+import org.apache.druid.segment.SimpleAscendingOffset;
+import org.apache.druid.segment.column.ValueType;
+import org.apache.druid.segment.data.GenericIndexed;
+import org.apache.druid.segment.data.VSizeColumnarInts;
+import org.apache.druid.segment.data.VSizeColumnarMultiInts;
+import org.apache.druid.segment.selector.TestColumnValueSelector;
+import org.apache.druid.segment.serde.DictionaryEncodedColumnSupplier;
+import org.apache.druid.testing.InitializedNullHandlingTest;
+import org.junit.Assert;
+import org.junit.Test;
+
+import javax.annotation.Nullable;
+import java.util.Arrays;
+
+public class PredicateValueMatcherFactoryTest extends InitializedNullHandlingTest
+{
+  @Test
+  public void testDefaultType()
+  {
+    Assert.assertEquals(ValueType.COMPLEX, forSelector(null).defaultType());
+  }
+
+  @Test
+  public void testDimensionProcessorSingleValuedDimensionMatchingValue()
+  {
+    final ValueMatcher matcher = forSelector("0").makeDimensionProcessor(DimensionSelector.constant("0"), false);
+    Assert.assertTrue(matcher.matches());
+  }
+
+  @Test
+  public void testDimensionProcessorSingleValuedDimensionNotMatchingValue()
+  {
+    final ValueMatcher matcher = forSelector("1").makeDimensionProcessor(DimensionSelector.constant("0"), false);
+    Assert.assertFalse(matcher.matches());
+  }
+
+  @Test
+  public void testDimensionProcessorMultiValuedDimensionMatchingValue()
+  {
+    // Emulate multi-valued dimension
+    final DictionaryEncodedColumnSupplier columnSupplier = new DictionaryEncodedColumnSupplier(
+        GenericIndexed.fromIterable(ImmutableList.of("v1", "v2", "v3"), GenericIndexed.STRING_STRATEGY),
+        null,
+        () -> VSizeColumnarMultiInts.fromIterable(ImmutableList.of(VSizeColumnarInts.fromArray(new int[]{1}))),
+        0
+    );
+    final ValueMatcher matcher = forSelector("v2")
+        .makeDimensionProcessor(columnSupplier.get().makeDimensionSelector(new SimpleAscendingOffset(1), null), true);
+    Assert.assertTrue(matcher.matches());
+  }
+
+  @Test
+  public void testDimensionProcessorMultiValuedDimensionNotMatchingValue()
+  {
+    // Emulate multi-valued dimension
+    final DictionaryEncodedColumnSupplier columnSupplier = new DictionaryEncodedColumnSupplier(
+        GenericIndexed.fromIterable(ImmutableList.of("v1", "v2", "v3"), GenericIndexed.STRING_STRATEGY),
+        null,
+        () -> VSizeColumnarMultiInts.fromIterable(ImmutableList.of(VSizeColumnarInts.fromArray(new int[]{1}))),
+        0
+    );
+    final ValueMatcher matcher = forSelector("v3")
+        .makeDimensionProcessor(columnSupplier.get().makeDimensionSelector(new SimpleAscendingOffset(1), null), true);
+    Assert.assertFalse(matcher.matches());
+  }
+
+  @Test
+  public void testFloatProcessorMatchingValue()
+  {
+    final TestColumnValueSelector<Float> columnValueSelector = TestColumnValueSelector.of(
+        Float.class,
+        ImmutableList.of(2.f),
+        DateTimes.nowUtc()
+    );
+    columnValueSelector.advance();
+    final ValueMatcher matcher = forSelector("2.f").makeFloatProcessor(columnValueSelector);
+    Assert.assertTrue(matcher.matches());
+  }
+
+  @Test
+  public void testFloatProcessorNotMatchingValue()
+  {
+    final TestColumnValueSelector<Float> columnValueSelector = TestColumnValueSelector.of(
+        Float.class,
+        ImmutableList.of(2.f),
+        DateTimes.nowUtc()
+    );
+    columnValueSelector.advance();
+    final ValueMatcher matcher = forSelector("5.f").makeFloatProcessor(columnValueSelector);
+    Assert.assertFalse(matcher.matches());
+  }
+
+  @Test
+  public void testDoubleProcessorMatchingValue()
+  {
+    final TestColumnValueSelector<Double> columnValueSelector = TestColumnValueSelector.of(
+        Double.class,
+        ImmutableList.of(2.),
+        DateTimes.nowUtc()
+    );
+    columnValueSelector.advance();
+    final ValueMatcher matcher = forSelector("2.").makeDoubleProcessor(columnValueSelector);
+    Assert.assertTrue(matcher.matches());
+  }
+
+  @Test
+  public void testDoubleProcessorNotMatchingValue()
+  {
+    final TestColumnValueSelector<Double> columnValueSelector = TestColumnValueSelector.of(
+        Double.class,
+        ImmutableList.of(2.),
+        DateTimes.nowUtc()
+    );
+    columnValueSelector.advance();
+    final ValueMatcher matcher = forSelector("5.").makeDoubleProcessor(columnValueSelector);
+    Assert.assertFalse(matcher.matches());
+  }
+
+  @Test
+  public void testLongProcessorMatchingValue()
+  {
+    final TestColumnValueSelector<Long> columnValueSelector = TestColumnValueSelector.of(
+        Long.class,
+        ImmutableList.of(2L),
+        DateTimes.nowUtc()
+    );
+    columnValueSelector.advance();
+    final ValueMatcher matcher = forSelector("2").makeLongProcessor(columnValueSelector);
+    Assert.assertTrue(matcher.matches());
+  }
+
+  @Test
+  public void testLongProcessorNotMatchingValue()
+  {
+    final TestColumnValueSelector<Long> columnValueSelector = TestColumnValueSelector.of(
+        Long.class,
+        ImmutableList.of(2L),
+        DateTimes.nowUtc()
+    );
+    columnValueSelector.advance();
+    final ValueMatcher matcher = forSelector("5").makeLongProcessor(columnValueSelector);
+    Assert.assertFalse(matcher.matches());
+  }
+
+  @Test
+  public void testComplexProcessorMatchingNull()
+  {
+    final TestColumnValueSelector<String> columnValueSelector = TestColumnValueSelector.of(
+        String.class,
+        Arrays.asList(null, "v"),
+        DateTimes.nowUtc()
+    );
+    columnValueSelector.advance();
+    final ValueMatcher matcher = forSelector(null).makeComplexProcessor(columnValueSelector);
+    Assert.assertTrue(matcher.matches());
+  }
+
+  @Test
+  public void testComplexProcessorEmptyString()
+  {
+    final TestColumnValueSelector<String> columnValueSelector = TestColumnValueSelector.of(
+        String.class,
+        Arrays.asList("", "v"),
+        DateTimes.nowUtc()
+    );
+    columnValueSelector.advance();
+    final ValueMatcher matcher = forSelector(null).makeComplexProcessor(columnValueSelector);
+    if (NullHandling.sqlCompatible()) {
+      Assert.assertFalse(matcher.matches());
+    } else {
+      Assert.assertTrue(matcher.matches());
+    }
+  }
+
+  @Test
+  public void testComplexProcessorMatchingInteger()
+  {
+    final TestColumnValueSelector<Integer> columnValueSelector = TestColumnValueSelector.of(
+        Integer.class,
+        ImmutableList.of(11),
+        DateTimes.nowUtc()
+    );
+    columnValueSelector.advance();
+    final ValueMatcher matcher = forSelector("11").makeComplexProcessor(columnValueSelector);
+    Assert.assertTrue(matcher.matches());
+  }
+
+  @Test
+  public void testComplexProcessorNotMatchingInteger()
+  {
+    final TestColumnValueSelector<Integer> columnValueSelector = TestColumnValueSelector.of(
+        Integer.class,
+        ImmutableList.of(15),
+        DateTimes.nowUtc()
+    );
+    columnValueSelector.advance();
+    final ValueMatcher matcher = forSelector("11").makeComplexProcessor(columnValueSelector);
+    Assert.assertFalse(matcher.matches());
+  }
+
+  @Test
+  public void testComplexProcessorMatchingLong()
+  {
+    final TestColumnValueSelector<Long> columnValueSelector = TestColumnValueSelector.of(
+        Long.class,
+        ImmutableList.of(11L),
+        DateTimes.nowUtc()
+    );
+    columnValueSelector.advance();
+    final ValueMatcher matcher = forSelector("11").makeComplexProcessor(columnValueSelector);
+    Assert.assertTrue(matcher.matches());
+  }
+
+  @Test
+  public void testComplexProcessorNotMatchingLong()
+  {
+    final TestColumnValueSelector<Long> columnValueSelector = TestColumnValueSelector.of(
+        Long.class,
+        ImmutableList.of(15L),
+        DateTimes.nowUtc()
+    );
+    columnValueSelector.advance();
+    final ValueMatcher matcher = forSelector("11").makeComplexProcessor(columnValueSelector);
+    Assert.assertFalse(matcher.matches());
+  }
+
+  @Test
+  public void testComplexProcessorMatchingFloat()
+  {
+    final TestColumnValueSelector<Float> columnValueSelector = TestColumnValueSelector.of(
+        Float.class,
+        ImmutableList.of(11.f),
+        DateTimes.nowUtc()
+    );
+    columnValueSelector.advance();
+    final ValueMatcher matcher = forSelector("11.f").makeComplexProcessor(columnValueSelector);
+    Assert.assertTrue(matcher.matches());
+  }
+
+  @Test
+  public void testComplexProcessorNotMatchingFloat()
+  {
+    final TestColumnValueSelector<Float> columnValueSelector = TestColumnValueSelector.of(
+        Float.class,
+        ImmutableList.of(15.f),
+        DateTimes.nowUtc()
+    );
+    columnValueSelector.advance();
+    final ValueMatcher matcher = forSelector("11.f").makeComplexProcessor(columnValueSelector);
+    Assert.assertFalse(matcher.matches());
+  }
+
+  @Test
+  public void testComplexProcessorMatchingDouble()
+  {
+    final TestColumnValueSelector<Double> columnValueSelector = TestColumnValueSelector.of(
+        Double.class,
+        ImmutableList.of(11.d),
+        DateTimes.nowUtc()
+    );
+    columnValueSelector.advance();
+    final ValueMatcher matcher = forSelector("11.d").makeComplexProcessor(columnValueSelector);
+    Assert.assertTrue(matcher.matches());
+  }
+
+  @Test
+  public void testComplexProcessorNotMatchingDouble()
+  {
+    final TestColumnValueSelector<Double> columnValueSelector = TestColumnValueSelector.of(
+        Double.class,
+        ImmutableList.of(15.d),
+        DateTimes.nowUtc()
+    );
+    columnValueSelector.advance();
+    final ValueMatcher matcher = forSelector("11.d").makeComplexProcessor(columnValueSelector);
+    Assert.assertFalse(matcher.matches());
+  }
+
+  @Test
+  public void testComplexProcessorMatchingString()
+  {
+    final TestColumnValueSelector<String> columnValueSelector = TestColumnValueSelector.of(
+        String.class,
+        ImmutableList.of("val"),
+        DateTimes.nowUtc()
+    );
+    columnValueSelector.advance();
+    final ValueMatcher matcher = forSelector("val").makeComplexProcessor(columnValueSelector);
+    Assert.assertTrue(matcher.matches());
+  }
+
+  @Test
+  public void testComplexProcessorNotMatchingString()
+  {
+    final TestColumnValueSelector<String> columnValueSelector = TestColumnValueSelector.of(
+        String.class,
+        ImmutableList.of("bar"),
+        DateTimes.nowUtc()
+    );
+    columnValueSelector.advance();
+    final ValueMatcher matcher = forSelector("val").makeComplexProcessor(columnValueSelector);
+    Assert.assertFalse(matcher.matches());
+  }
+
+  @Test
+  public void testComplexProcessorMatchingStringList()
+  {
+    final TestColumnValueSelector<String> columnValueSelector = TestColumnValueSelector.of(
+        String.class,
+        ImmutableList.of(ImmutableList.of("val")),
+        DateTimes.nowUtc()
+    );
+    columnValueSelector.advance();
+    final ValueMatcher matcher = forSelector("val").makeComplexProcessor(columnValueSelector);
+    Assert.assertTrue(matcher.matches());
+  }
+
+  @Test
+  public void testComplexProcessorNotMatchingStringList()
+  {
+    final TestColumnValueSelector<String> columnValueSelector = TestColumnValueSelector.of(
+        String.class,
+        ImmutableList.of(ImmutableList.of("bar")),
+        DateTimes.nowUtc()
+    );
+    columnValueSelector.advance();
+    final ValueMatcher matcher = forSelector("val").makeComplexProcessor(columnValueSelector);
+    Assert.assertFalse(matcher.matches());
+  }
+
+  @Test
+  public void testComplexProcessorMatchingEmptyList()
+  {
+    final TestColumnValueSelector<String> columnValueSelector = TestColumnValueSelector.of(
+        String.class,
+        ImmutableList.of(ImmutableList.of()),
+        DateTimes.nowUtc()
+    );
+    columnValueSelector.advance();
+    final ValueMatcher matcher = forSelector(null).makeComplexProcessor(columnValueSelector);
+    Assert.assertTrue(matcher.matches());
+  }
+
+  @Test
+  public void testComplexProcessorMatchingBoolean()
+  {
+    final TestColumnValueSelector<String> columnValueSelector = TestColumnValueSelector.of(
+        String.class,
+        ImmutableList.of(false),
+        DateTimes.nowUtc()
+    );
+    columnValueSelector.advance();
+    final ValueMatcher matcher = forSelector("false").makeComplexProcessor(columnValueSelector);
+    Assert.assertTrue(matcher.matches());
+  }
+
+  @Test
+  public void testComplexProcessorNotMatchingBoolean()
+  {
+    final TestColumnValueSelector<String> columnValueSelector = TestColumnValueSelector.of(
+        String.class,
+        ImmutableList.of(true),
+        DateTimes.nowUtc()
+    );
+    columnValueSelector.advance();
+    final ValueMatcher matcher = forSelector("false").makeComplexProcessor(columnValueSelector);
+    Assert.assertFalse(matcher.matches());
+  }
+
+  @Test
+  public void testComplexProcessorMatchingByteArray()
+  {
+    final TestColumnValueSelector<String> columnValueSelector = TestColumnValueSelector.of(
+        String.class,
+        ImmutableList.of(StringUtils.toUtf8("var")),
+        DateTimes.nowUtc()
+    );
+    columnValueSelector.advance();
+    final String base64Encoded = StringUtils.encodeBase64String(StringUtils.toUtf8("var"));
+    final ValueMatcher matcher = forSelector(base64Encoded).makeComplexProcessor(columnValueSelector);
+    Assert.assertTrue(matcher.matches());
+  }
+
+  @Test
+  public void testComplexProcessorNotMatchingByteArray()
+  {
+    final TestColumnValueSelector<String> columnValueSelector = TestColumnValueSelector.of(
+        String.class,
+        ImmutableList.of(StringUtils.toUtf8("var")),
+        DateTimes.nowUtc()
+    );
+    columnValueSelector.advance();
+    final ValueMatcher matcher = forSelector("val").makeComplexProcessor(columnValueSelector);
+    Assert.assertFalse(matcher.matches());
+  }
+
+  private static PredicateValueMatcherFactory forSelector(@Nullable String value)
+  {
+    return new PredicateValueMatcherFactory(new SelectorPredicateFactory(value));
+  }
+}

processing/src/test/java/org/apache/druid/segment/selector/TestColumnValueSelector.java

@@ -0,0 +1,179 @@
+/*
+ * 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.segment.selector;
+
+import org.apache.druid.query.dimension.DimensionSpec;
+import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector;
+import org.apache.druid.segment.ColumnSelectorFactory;
+import org.apache.druid.segment.ColumnValueSelector;
+import org.apache.druid.segment.Cursor;
+import org.apache.druid.segment.DimensionSelector;
+import org.apache.druid.segment.column.ColumnCapabilities;
+import org.joda.time.DateTime;
+
+import javax.annotation.Nullable;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.function.Supplier;
+import java.util.stream.Stream;
+
+public class TestColumnValueSelector<T> implements ColumnValueSelector<Object>, Cursor
+{
+  private final Class<T> clazz;
+  private final Supplier<Iterator<Object>> iteratorSupplier;
+  private final DateTime time;
+
+  private Iterator<Object> iterator;
+  private Object value;
+
+  public static <T> TestColumnValueSelector<T> of(Class<T> clazz, Collection<Object> collection, DateTime time)
+  {
+    return new TestColumnValueSelector<>(clazz, collection::iterator, time);
+  }
+
+  public static <T> TestColumnValueSelector<T> of(Class<T> clazz, Stream<Object> stream, DateTime time)
+  {
+    return new TestColumnValueSelector<>(clazz, stream::iterator, time);
+  }
+
+  protected TestColumnValueSelector(Class<T> clazz, Supplier<Iterator<Object>> iteratorSupplier, DateTime time)
+  {
+    this.clazz = clazz;
+    this.iteratorSupplier = iteratorSupplier;
+    this.time = time;
+    this.iterator = iteratorSupplier.get();
+  }
+
+  @Override
+  public ColumnSelectorFactory getColumnSelectorFactory()
+  {
+    return new ColumnSelectorFactory()
+    {
+      @Override
+      public DimensionSelector makeDimensionSelector(DimensionSpec dimensionSpec)
+      {
+        throw new UnsupportedOperationException("Not implemented");
+      }
+
+      @Override
+      public ColumnValueSelector makeColumnValueSelector(String columnName)
+      {
+        return TestColumnValueSelector.this;
+      }
+
+      @Nullable
+      @Override
+      public ColumnCapabilities getColumnCapabilities(String column)
+      {
+        return null;
+      }
+    };
+  }
+
+  @Override
+  public DateTime getTime()
+  {
+    return time;
+  }
+
+  @Override
+  public void advance()
+  {
+    value = iterator.next();
+  }
+
+  @Override
+  public void advanceUninterruptibly()
+  {
+    advance();
+  }
+
+  @Override
+  public boolean isDone()
+  {
+    return !iterator.hasNext();
+  }
+
+  @Override
+  public boolean isDoneOrInterrupted()
+  {
+    return isDone();
+  }
+
+  @Override
+  public void reset()
+  {
+    iterator = iteratorSupplier.get();
+  }
+
+  @Override
+  public double getDouble()
+  {
+    if (value instanceof Number) {
+      return ((Number) value).doubleValue();
+    } else {
+      return Double.parseDouble(value.toString());
+    }
+  }
+
+  @Override
+  public float getFloat()
+  {
+    if (value instanceof Number) {
+      return ((Number) value).floatValue();
+    } else {
+      return Float.parseFloat(value.toString());
+    }
+  }
+
+  @Override
+  public long getLong()
+  {
+    if (value instanceof Number) {
+      return ((Number) value).longValue();
+    } else {
+      return Long.parseLong(value.toString());
+    }
+  }
+
+  @Override
+  public void inspectRuntimeShape(RuntimeShapeInspector inspector)
+  {
+  }
+
+  @Override
+  public boolean isNull()
+  {
+    return value == null;
+  }
+
+  @Nullable
+  @Override
+  public Object getObject()
+  {
+    return value;
+  }
+
+  @Override
+  public Class<? extends T> classOfObject()
+  {
+    return clazz;
+  }
+}

processing/src/test/java/org/apache/druid/segment/transform/TransformerTest.java

@@ -0,0 +1,220 @@
+/*
+ * 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.segment.transform;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.InputRowListPlusRawValues;
+import org.apache.druid.data.input.MapBasedInputRow;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.query.expression.TestExprMacroTable;
+import org.apache.druid.query.filter.SelectorDimFilter;
+import org.apache.druid.testing.InitializedNullHandlingTest;
+import org.joda.time.DateTime;
+import org.junit.Assert;
+import org.junit.Ignore;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+
+public class TransformerTest extends InitializedNullHandlingTest
+{
+  @Rule
+  public ExpectedException expectedException = ExpectedException.none();
+
+  @Test
+  public void testTransformNullRowReturnNull()
+  {
+    final Transformer transformer = new Transformer(new TransformSpec(null, null));
+    Assert.assertNull(transformer.transform((InputRow) null));
+    Assert.assertNull(transformer.transform((InputRowListPlusRawValues) null));
+  }
+
+  @Test
+  public void testTransformTimeColumn()
+  {
+    final Transformer transformer = new Transformer(
+        new TransformSpec(
+            null,
+            ImmutableList.of(
+                new ExpressionTransform("__time", "timestamp_shift(__time, 'P1D', -2)", TestExprMacroTable.INSTANCE)
+            )
+        )
+    );
+    final DateTime now = DateTimes.nowUtc();
+    final InputRow row = new MapBasedInputRow(
+        now,
+        ImmutableList.of("dim"),
+        ImmutableMap.of("__time", now, "dim", false)
+    );
+    final InputRow actual = transformer.transform(row);
+    Assert.assertNotNull(actual);
+    Assert.assertEquals(now.minusDays(2), actual.getTimestamp());
+  }
+
+  @Test
+  public void testTransformWithStringTransformOnBooleanColumnTransformAfterCasting()
+  {
+    final Transformer transformer = new Transformer(
+        new TransformSpec(
+            null,
+            ImmutableList.of(new ExpressionTransform("dim", "strlen(dim)", TestExprMacroTable.INSTANCE))
+        )
+    );
+    final InputRow row = new MapBasedInputRow(
+        DateTimes.nowUtc(),
+        ImmutableList.of("dim"),
+        ImmutableMap.of("dim", false)
+    );
+    final InputRow actual = transformer.transform(row);
+    Assert.assertNotNull(actual);
+    Assert.assertEquals(ImmutableList.of("dim"), actual.getDimensions());
+    Assert.assertEquals(5L, actual.getRaw("dim"));
+    Assert.assertEquals(row.getTimestamp(), actual.getTimestamp());
+  }
+
+  @Test
+  public void testTransformWithStringTransformOnLongColumnTransformAfterCasting()
+  {
+    final Transformer transformer = new Transformer(
+        new TransformSpec(
+            null,
+            ImmutableList.of(new ExpressionTransform("dim", "strlen(dim)", TestExprMacroTable.INSTANCE))
+        )
+    );
+    final InputRow row = new MapBasedInputRow(
+        DateTimes.nowUtc(),
+        ImmutableList.of("dim"),
+        ImmutableMap.of("dim", 10L)
+    );
+    final InputRow actual = transformer.transform(row);
+    Assert.assertNotNull(actual);
+    Assert.assertEquals(ImmutableList.of("dim"), actual.getDimensions());
+    Assert.assertEquals(2L, actual.getRaw("dim"));
+    Assert.assertEquals(row.getTimestamp(), actual.getTimestamp());
+  }
+
+  @Test
+  public void testTransformWithStringTransformOnDoubleColumnTransformAfterCasting()
+  {
+    final Transformer transformer = new Transformer(
+        new TransformSpec(
+            null,
+            ImmutableList.of(new ExpressionTransform("dim", "strlen(dim)", TestExprMacroTable.INSTANCE))
+        )
+    );
+    final InputRow row = new MapBasedInputRow(
+        DateTimes.nowUtc(),
+        ImmutableList.of("dim"),
+        ImmutableMap.of("dim", 200.5d)
+    );
+    final InputRow actual = transformer.transform(row);
+    Assert.assertNotNull(actual);
+    Assert.assertEquals(ImmutableList.of("dim"), actual.getDimensions());
+    Assert.assertEquals(5L, actual.getRaw("dim"));
+    Assert.assertEquals(row.getTimestamp(), actual.getTimestamp());
+  }
+
+  @Ignore("Disabled until https://github.com/apache/druid/issues/9824 is fixed")
+  @Test
+  public void testTransformWithStringTransformOnListColumnThrowingException()
+  {
+    final Transformer transformer = new Transformer(
+        new TransformSpec(
+            null,
+            ImmutableList.of(new ExpressionTransform("dim", "strlen(dim)", TestExprMacroTable.INSTANCE))
+        )
+    );
+    final InputRow row = new MapBasedInputRow(
+        DateTimes.nowUtc(),
+        ImmutableList.of("dim"),
+        ImmutableMap.of("dim", ImmutableList.of(10, 20, 100))
+    );
+    final InputRow actual = transformer.transform(row);
+    Assert.assertNotNull(actual);
+    Assert.assertEquals(ImmutableList.of("dim"), actual.getDimensions());
+    // Unlike for querying, Druid doesn't explode multi-valued columns automatically for ingestion.
+    expectedException.expect(AssertionError.class);
+    actual.getRaw("dim");
+  }
+
+  @Test
+  public void testTransformWithSelectorFilterWithStringBooleanValueOnBooleanColumnFilterAfterCasting()
+  {
+    final Transformer transformer = new Transformer(
+        new TransformSpec(new SelectorDimFilter("dim", "false", null), null)
+    );
+    final InputRow row1 = new MapBasedInputRow(
+        DateTimes.nowUtc(),
+        ImmutableList.of("dim"),
+        ImmutableMap.of("dim", false)
+    );
+    Assert.assertEquals(row1, transformer.transform(row1));
+    final InputRow row2 = new MapBasedInputRow(
+        DateTimes.nowUtc(),
+        ImmutableList.of("dim"),
+        ImmutableMap.of("dim", true)
+    );
+    Assert.assertNull(transformer.transform(row2));
+  }
+
+  @Test
+  public void testTransformWithSelectorFilterWithStringBooleanValueOnStringColumn()
+  {
+    final Transformer transformer = new Transformer(
+        new TransformSpec(new SelectorDimFilter("dim", "false", null), null)
+    );
+    final InputRow row = new MapBasedInputRow(
+        DateTimes.nowUtc(),
+        ImmutableList.of("dim"),
+        ImmutableMap.of("dim", "false")
+    );
+    Assert.assertEquals(row, transformer.transform(row));
+    final InputRow row2 = new MapBasedInputRow(
+        DateTimes.nowUtc(),
+        ImmutableList.of("dim"),
+        ImmutableMap.of("dim", "true")
+    );
+    Assert.assertNull(transformer.transform(row2));
+  }
+
+  @Test
+  public void testTransformWithTransformAndFilterTransformFirst()
+  {
+    final Transformer transformer = new Transformer(
+        new TransformSpec(
+            new SelectorDimFilter("dim", "0", null),
+            // A boolean expression returns a long.
+            ImmutableList.of(new ExpressionTransform("dim", "strlen(dim) == 10", TestExprMacroTable.INSTANCE))
+        )
+    );
+    final InputRow row = new MapBasedInputRow(
+        DateTimes.nowUtc(),
+        ImmutableList.of("dim"),
+        ImmutableMap.of("dim", "short")
+    );
+    final InputRow actual = transformer.transform(row);
+    Assert.assertNotNull(actual);
+    Assert.assertEquals(ImmutableList.of("dim"), actual.getDimensions());
+    Assert.assertEquals(0L, actual.getRaw("dim"));
+    Assert.assertEquals(row.getTimestamp(), actual.getTimestamp());
+  }
+}