Fixing an issue with filtering on a single dimension by converting In… (#14277)

* Fixing an issue with filtering on a single dimension by converting In filter to a selector filter as needed with Filters.toFilter

* Adding a test so that any future refactoring does not break this behavior

* Made comment a bit more meaningful

processing/src/main/java/org/apache/druid/segment/join/JoinableFactoryWrapper.java

@@ -30,6 +30,7 @@ import org.apache.druid.java.util.common.Pair;
 import org.apache.druid.query.filter.Filter;
 import org.apache.druid.query.filter.InDimFilter;
 import org.apache.druid.segment.filter.FalseFilter;
+import org.apache.druid.segment.filter.Filters;
 import org.apache.druid.utils.CollectionUtils;
 
 import javax.annotation.Nullable;
@@ -177,7 +178,10 @@ public class JoinableFactoryWrapper
         }
         return new JoinClauseToFilterConversion(null, false);
       }
-      final Filter onlyFilter = new InDimFilter(leftColumn, columnValuesWithUniqueFlag.getColumnValues());
+      final Filter onlyFilter = Filters.toFilter(new InDimFilter(
+          leftColumn,
+          columnValuesWithUniqueFlag.getColumnValues()
+      ));
       if (!columnValuesWithUniqueFlag.isAllUnique()) {
         joinClauseFullyConverted = false;
       }

processing/src/test/java/org/apache/druid/segment/join/JoinableFactoryWrapperTest.java

@@ -36,6 +36,7 @@ import org.apache.druid.query.filter.InDimFilter;
 import org.apache.druid.segment.column.ColumnType;
 import org.apache.druid.segment.column.RowSignature;
 import org.apache.druid.segment.filter.FalseFilter;
+import org.apache.druid.segment.filter.SelectorFilter;
 import org.apache.druid.segment.join.lookup.LookupJoinable;
 import org.apache.druid.segment.join.table.IndexedTable;
 import org.apache.druid.segment.join.table.IndexedTableJoinable;
@@ -100,6 +101,13 @@ public class JoinableFactoryWrapperTest extends NullHandlingTest
                   .build()
   );
 
+  private static final InlineDataSource INDEXED_TABLE_DS_ONE_ROW = InlineDataSource.fromIterable(
+      ImmutableList.of(
+          new Object[]{"Mexico"}
+      ),
+      RowSignature.builder().add("country", ColumnType.STRING).build()
+  );
+
   private static final InlineDataSource NULL_INDEXED_TABLE_DS = InlineDataSource.fromIterable(
       ImmutableList.of(
           new Object[]{null}
@@ -123,6 +131,14 @@ public class JoinableFactoryWrapperTest extends NullHandlingTest
       DateTimes.nowUtc().toString()
   );
 
+  private static final IndexedTable TEST_INDEXED_TABLE_ONE_ROW = new RowBasedIndexedTable<>(
+      INDEXED_TABLE_DS_ONE_ROW.getRowsAsList(),
+      INDEXED_TABLE_DS_ONE_ROW.rowAdapter(),
+      INDEXED_TABLE_DS_ONE_ROW.getRowSignature(),
+      ImmutableSet.of("country"),
+      DateTimes.nowUtc().toString()
+  );
+
   private static final IndexedTable TEST_NULL_INDEXED_TABLE = new RowBasedIndexedTable<>(
       NULL_INDEXED_TABLE_DS.getRowsAsList(),
       NULL_INDEXED_TABLE_DS.rowAdapter(),
@@ -240,6 +256,33 @@ public class JoinableFactoryWrapperTest extends NullHandlingTest
     );
   }
 
+  @Test
+  public void test_convertJoinsToPartialFiltersWithSelectorFiltersInsteadOfInsForSingleValue()
+  {
+    JoinableClause joinableClause = new JoinableClause(
+        "j.",
+        new IndexedTableJoinable(TEST_INDEXED_TABLE_ONE_ROW),
+        JoinType.INNER,
+        JoinConditionAnalysis.forExpression("x == \"j.country\"", "j.", ExprMacroTable.nil())
+    );
+    final Pair<List<Filter>, List<JoinableClause>> conversion = JoinableFactoryWrapper.convertJoinsToFilters(
+        ImmutableList.of(joinableClause),
+        ImmutableSet.of("x"),
+        Integer.MAX_VALUE
+    );
+
+    // Although the filter created was an In Filter in equijoin (here inFilter = IN (Mexico))
+    // We should receive a SelectorFilter for Filters.toFilter(inFilter) call
+    // and should receive a SelectorFilter with x = Mexico
+    Assert.assertEquals(
+        Pair.of(
+            ImmutableList.of(new SelectorFilter("x", "Mexico", null)),
+            ImmutableList.of()
+        ),
+        conversion
+    );
+  }
+
   @Test
   public void test_convertJoinsToFilters_convertTwoInnerJoins()
   {