Allow typedIn to run in replace-with-default mode. (#16233)

* Allow typedIn to run in replace-with-default mode.

Useful when data servers, like Historicals, are running in replace-with-default
mode and the Broker is running in SQL-compatible mode, which can happen during
a rolling update that is applying a mode change.

processing/src/main/java/org/apache/druid/query/filter/TypedInFilter.java

@@ -223,7 +223,9 @@ public class TypedInFilter extends AbstractOptimizableDimFilter implements Filte
   @Override
   public DimFilter optimize(final boolean mayIncludeUnknown)
   {
-    checkSqlCompatible();
+    if (NullHandling.replaceWithDefault()) {
+      return convertToLegacy().optimize(mayIncludeUnknown);
+    }
     final List<?> matchValues = this.sortedMatchValues.get();
     if (matchValues.isEmpty()) {
       return FalseDimFilter.instance();
@@ -244,7 +246,9 @@ public class TypedInFilter extends AbstractOptimizableDimFilter implements Filte
   @Override
   public Filter toFilter()
   {
-    checkSqlCompatible();
+    if (NullHandling.replaceWithDefault()) {
+      return convertToLegacy().toFilter();
+    }
     return this;
   }
 
@@ -417,13 +421,14 @@ public class TypedInFilter extends AbstractOptimizableDimFilter implements Filte
         .build();
   }
 
-  private void checkSqlCompatible()
+  private InDimFilter convertToLegacy()
   {
-    if (NullHandling.replaceWithDefault()) {
-      throw InvalidInput.exception(
-          "Invalid IN filter, typed in filter only supports SQL compatible null handling mode, set druid.generic.useDefaultValue=false to use this filter"
-      );
-    }
+    return new InDimFilter(
+      column,
+      InDimFilter.ValuesSet.copyOf(sortedMatchValues.get().stream().map(Evals::asString).iterator()),
+      null,
+      filterTuning
+    );
   }
 
   private static boolean checkSorted(List<?> unsortedValues, ColumnType matchValueType)

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

@@ -627,25 +627,16 @@ public class InFilterTests
 
     private void assertTypedFilterMatches(DimFilter filter, List<String> expectedRows)
     {
-      // this filter only tests in sql compatible mode
-      if (NullHandling.sqlCompatible()) {
-        super.assertFilterMatches(filter, expectedRows);
-        try {
-          // make sure round trip json serde is cool
-          super.assertFilterMatches(
-              jsonMapper.readValue(jsonMapper.writeValueAsString(filter), DimFilter.class),
-              expectedRows
-          );
-        }
-        catch (JsonProcessingException e) {
-          throw new RuntimeException(e);
-        }
-      } else {
-        Throwable t = Assert.assertThrows(
-            DruidException.class,
-            () -> super.assertFilterMatches(filter, expectedRows)
+      super.assertFilterMatches(filter, expectedRows);
+      try {
+        // make sure round trip json serde is cool
+        super.assertFilterMatches(
+            jsonMapper.readValue(jsonMapper.writeValueAsString(filter), DimFilter.class),
+            expectedRows
         );
-        Assert.assertEquals("Invalid IN filter, typed in filter only supports SQL compatible null handling mode, set druid.generic.useDefaultValue=false to use this filter", t.getMessage());
+      }
+      catch (JsonProcessingException e) {
+        throw new RuntimeException(e);
       }
     }
 
@@ -718,6 +709,15 @@ public class InFilterTests
       Assert.assertEquals(filter, mapper.readValue(s, TypedInFilter.class));
     }
 
+    @Test
+    public void testConvertToLegacy()
+    {
+      Assume.assumeTrue(NullHandling.replaceWithDefault());
+      TypedInFilter aFilter = new TypedInFilter("column", ColumnType.STRING, Arrays.asList("a", "b", "c"), null, null);
+      Assert.assertTrue(aFilter.optimize(false) instanceof InDimFilter);
+      Assert.assertTrue(aFilter.toFilter() instanceof InDimFilter);
+    }
+
     @Test
     public void testGetCacheKey()
     {
@@ -786,14 +786,6 @@ public class InFilterTests
     @Test
     public void testInvalidParameters()
     {
-      if (NullHandling.replaceWithDefault()) {
-        Throwable t = Assert.assertThrows(
-            DruidException.class,
-            () -> new TypedInFilter("column", ColumnType.STRING, Collections.emptyList(), null, null).toFilter()
-        );
-        Assert.assertEquals("Invalid IN filter, typed in filter only supports SQL compatible null handling mode, set druid.generic.useDefaultValue=false to use this filter", t.getMessage());
-      }
-
       Assume.assumeTrue(NullHandling.sqlCompatible());
       Throwable t = Assert.assertThrows(
           DruidException.class,