Add support for raw JSON filter predicates

pinot-core/src/main/java/org/apache/pinot/core/operator/filter/predicate/EqualsPredicateEvaluatorFactory.java

@@ -80,6 +80,7 @@ public static EqRawPredicateEvaluator newRawValueBasedEvaluator(EqPredicate eqPr
       case TIMESTAMP:
         return new LongRawValueBasedEqPredicateEvaluator(eqPredicate, TimestampUtils.toMillisSinceEpoch(value));
       case STRING:
+      case JSON:
         return new StringRawValueBasedEqPredicateEvaluator(eqPredicate, value);
       case BYTES:
         return new BytesRawValueBasedEqPredicateEvaluator(eqPredicate, BytesUtils.toBytes(value));

pinot-core/src/main/java/org/apache/pinot/core/operator/filter/predicate/InPredicateEvaluatorFactory.java

@@ -128,7 +128,8 @@ public static InRawPredicateEvaluator newRawValueBasedEvaluator(InPredicate inPr
         }
         return new LongRawValueBasedInPredicateEvaluator(inPredicate, matchingValues);
       }
-      case STRING: {
+      case STRING:
+      case JSON: {
         List<String> stringValues = inPredicate.getValues();
         Set<String> matchingValues = new ObjectOpenHashSet<>(HashUtil.getMinHashSetSize(stringValues.size()));
         // NOTE: Add value-by-value to avoid overhead

pinot-core/src/main/java/org/apache/pinot/core/operator/filter/predicate/NotEqualsPredicateEvaluatorFactory.java

@@ -76,6 +76,7 @@ public static NeqRawPredicateEvaluator newRawValueBasedEvaluator(NotEqPredicate
       case TIMESTAMP:
         return new LongRawValueBasedNeqPredicateEvaluator(notEqPredicate, TimestampUtils.toMillisSinceEpoch(value));
       case STRING:
+      case JSON:
         return new StringRawValueBasedNeqPredicateEvaluator(notEqPredicate, value);
       case BYTES:
         return new BytesRawValueBasedNeqPredicateEvaluator(notEqPredicate, BytesUtils.toBytes(value));

pinot-core/src/main/java/org/apache/pinot/core/operator/filter/predicate/NotInPredicateEvaluatorFactory.java

@@ -128,7 +128,8 @@ public static NotInRawPredicateEvaluator newRawValueBasedEvaluator(NotInPredicat
         }
         return new LongRawValueBasedNotInPredicateEvaluator(notInPredicate, nonMatchingValues);
       }
-      case STRING: {
+      case STRING:
+      case JSON: {
         List<String> stringValues = notInPredicate.getValues();
         Set<String> nonMatchingValues = new ObjectOpenHashSet<>(HashUtil.getMinHashSetSize(stringValues.size()));
         // NOTE: Add value-by-value to avoid overhead

pinot-core/src/test/java/org/apache/pinot/core/operator/filter/predicate/NoDictionaryEqualsPredicateEvaluatorsTest.java

@@ -248,6 +248,46 @@ public void testStringPredicateEvaluators() {
     }
   }
 
+  @Test
+  public void testJsonPredicateEvaluators() {
+    String jsonStringTemplate = "{\"id\": %s, \"name\": %s}";
+    String jsonString = String.format(jsonStringTemplate,
+        RandomStringUtils.randomAlphanumeric(MAX_STRING_LENGTH),
+        RandomStringUtils.randomAlphanumeric(MAX_STRING_LENGTH));
+
+    EqPredicate eqPredicate = new EqPredicate(COLUMN_EXPRESSION, jsonString);
+    PredicateEvaluator eqPredicateEvaluator =
+        EqualsPredicateEvaluatorFactory.newRawValueBasedEvaluator(eqPredicate, FieldSpec.DataType.STRING);
+
+    NotEqPredicate notEqPredicate = new NotEqPredicate(COLUMN_EXPRESSION, jsonString);
+    PredicateEvaluator neqPredicateEvaluator =
+        NotEqualsPredicateEvaluatorFactory.newRawValueBasedEvaluator(notEqPredicate, FieldSpec.DataType.STRING);
+
+    Assert.assertTrue(eqPredicateEvaluator.applySV(jsonString));
+    Assert.assertFalse(neqPredicateEvaluator.applySV(jsonString));
+
+    String[] randomJsonStrings = new String[NUM_MULTI_VALUES];
+    PredicateEvaluatorTestUtils.fillRandomJson(randomJsonStrings, jsonStringTemplate, 2, MAX_STRING_LENGTH);
+    randomJsonStrings[_random.nextInt(NUM_MULTI_VALUES)] = jsonString;
+
+    Assert.assertTrue(eqPredicateEvaluator.applyMV(randomJsonStrings, NUM_MULTI_VALUES));
+    Assert.assertFalse(neqPredicateEvaluator.applyMV(randomJsonStrings, NUM_MULTI_VALUES));
+
+    for (int i = 0; i < 100; i++) {
+      String randomJson = String.format(jsonStringTemplate,
+          RandomStringUtils.randomAlphanumeric(MAX_STRING_LENGTH),
+          RandomStringUtils.randomAlphanumeric(MAX_STRING_LENGTH));
+      Assert.assertEquals(eqPredicateEvaluator.applySV(randomJson), (randomJson.equals(jsonString)));
+      Assert.assertEquals(neqPredicateEvaluator.applySV(randomJson), (!randomJson.equals(jsonString)));
+
+      PredicateEvaluatorTestUtils.fillRandomJson(randomJsonStrings, jsonStringTemplate, 2, MAX_STRING_LENGTH);
+      Assert.assertEquals(eqPredicateEvaluator.applyMV(randomJsonStrings, NUM_MULTI_VALUES),
+          ArrayUtils.contains(randomJsonStrings, jsonString));
+      Assert.assertEquals(neqPredicateEvaluator.applyMV(randomJsonStrings, NUM_MULTI_VALUES),
+          !ArrayUtils.contains(randomJsonStrings, jsonString));
+    }
+  }
+
   @Test
   public void testBytesPredicateEvaluators() {
     byte[] bytesValue = RandomStringUtils.random(MAX_STRING_LENGTH).getBytes();

pinot-core/src/test/java/org/apache/pinot/core/operator/filter/predicate/NoDictionaryInPredicateEvaluatorTest.java

@@ -283,6 +283,50 @@ public void testStringPredicateEvaluators() {
     Assert.assertFalse(notInPredicateEvaluator.applyMV(multiValues, NUM_MULTI_VALUES));
   }
 
+  @Test
+  public void testJsonPredicateEvaluators() {
+    String jsonStringTemplate = "{\"id\": %s, \"name\": %s}";
+
+    List<String> jsonValues = new ArrayList<>(NUM_PREDICATE_VALUES);
+    Set<String> jsonValueSet = new HashSet<>();
+
+    for (int i = 0; i < 100; i++) {
+      String jsonString = String.format(jsonStringTemplate,
+          RandomStringUtils.randomAlphanumeric(MAX_STRING_LENGTH),
+          RandomStringUtils.randomAlphanumeric(MAX_STRING_LENGTH));
+      jsonValues.add(jsonString);
+      jsonValueSet.add(jsonString);
+    }
+
+    InPredicate inPredicate = new InPredicate(COLUMN_EXPRESSION, jsonValues);
+    PredicateEvaluator inPredicateEvaluator =
+        InPredicateEvaluatorFactory.newRawValueBasedEvaluator(inPredicate, FieldSpec.DataType.STRING);
+
+    NotInPredicate notInPredicate = new NotInPredicate(COLUMN_EXPRESSION, jsonValues);
+    PredicateEvaluator notInPredicateEvaluator =
+        NotInPredicateEvaluatorFactory.newRawValueBasedEvaluator(notInPredicate, FieldSpec.DataType.STRING);
+
+    for (String value : jsonValueSet) {
+      Assert.assertTrue(inPredicateEvaluator.applySV(value));
+      Assert.assertFalse(notInPredicateEvaluator.applySV(value));
+    }
+
+    for (int i = 0; i < 100; i++) {
+      String randomJsonValue = String.format(jsonStringTemplate,
+          RandomStringUtils.randomAlphanumeric(MAX_STRING_LENGTH),
+          RandomStringUtils.randomAlphanumeric(MAX_STRING_LENGTH));
+      Assert.assertEquals(inPredicateEvaluator.applySV(randomJsonValue), jsonValueSet.contains(randomJsonValue));
+      Assert.assertEquals(notInPredicateEvaluator.applySV(randomJsonValue), !jsonValueSet.contains(randomJsonValue));
+    }
+
+    String[] multiValues = new String[NUM_MULTI_VALUES];
+    PredicateEvaluatorTestUtils.fillRandomJson(multiValues, jsonStringTemplate, 2, MAX_STRING_LENGTH);
+    multiValues[_random.nextInt(NUM_MULTI_VALUES)] = jsonValues.get(_random.nextInt(NUM_PREDICATE_VALUES));
+
+    Assert.assertTrue(inPredicateEvaluator.applyMV(multiValues, NUM_MULTI_VALUES));
+    Assert.assertFalse(notInPredicateEvaluator.applyMV(multiValues, NUM_MULTI_VALUES));
+  }
+
   @Test
   public void testBytesPredicateEvaluators() {
     List<String> stringValues = new ArrayList<>(NUM_PREDICATE_VALUES);

pinot-core/src/test/java/org/apache/pinot/core/operator/filter/predicate/PredicateEvaluatorTestUtils.java

@@ -70,4 +70,15 @@ public static void fillRandom(byte[][] randomValues, int maxStringLength) {
       randomValues[i] = RandomStringUtils.random(maxStringLength).getBytes();
     }
   }
+
+  public static void fillRandomJson(String[] randomValues, String jsonStringTemplate, int numPlaceholders,
+      int maxStringLength) {
+    for (int i = 0; i < randomValues.length; i++) {
+      Object[] randomPlaceholderValues = new String[numPlaceholders];
+      for (int j = 0; j < numPlaceholders; j++) {
+        randomPlaceholderValues[j] = RandomStringUtils.randomAlphanumeric(maxStringLength);
+      }
+      randomValues[i] = String.format(jsonStringTemplate, randomPlaceholderValues);
+    }
+  }
 }