Rebased to latest

Signed-off-by: Krishna Kondaka <[email protected]>

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/event/DataType.java

@@ -75,4 +75,24 @@ public String getTypeName() {
     static DataType fromTypeName(final String option) {
         return TYPES_MAP.get(option);
     }
+
+    public static boolean isSameType(final Object object, final String option) {
+        DataType type = fromTypeName(option);
+        if (type == null)
+            return false;
+        switch (type) {
+          case STRING:
+            return (object instanceof String);
+          case BOOLEAN:
+            return (object instanceof Boolean);
+          case INTEGER:
+            return (object instanceof Integer);
+          case LONG:
+            return (object instanceof Long);
+          case DOUBLE:
+            return (object instanceof Double);
+          default:
+            return false;
+        }
+    }
 }

data-prepper-expression/src/main/antlr/DataPrepperExpression.g4

@@ -78,7 +78,9 @@ regexEqualityOperator
 
 relationalOperatorExpression
     : relationalOperatorExpression relationalOperator setOperatorExpression
+    | relationalOperatorExpression relationalOperator typeOfOperatorExpression
     | setOperatorExpression
+    | typeOfOperatorExpression
     ;
 
 relationalOperator
@@ -88,6 +90,10 @@ relationalOperator
     | GTE
     ;
 
+typeOfOperatorExpression
+    : JsonPointer TYPEOF String
+    ;
+
 setOperatorExpression
     : setOperatorExpression setOperator setInitializer
     | unaryOperatorExpression
@@ -298,6 +304,7 @@ MATCH_REGEX_PATTERN : '=~';
 NOT_MATCH_REGEX_PATTERN : '!~';
 IN_SET : SPACE 'in' SPACE;
 NOT_IN_SET : SPACE 'not in' SPACE;
+TYPEOF: SPACE 'typeof' SPACE;
 AND : SPACE 'and' SPACE;
 OR : SPACE 'or' SPACE;
 NOT : 'not' SPACE;

data-prepper-expression/src/main/java/org/opensearch/dataprepper/expression/GenericTypeOfOperator.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.expression;
+
+import org.antlr.v4.runtime.RuleContext;
+import org.opensearch.dataprepper.expression.antlr.DataPrepperExpressionParser;
+
+import java.util.Set;
+import java.util.function.BiPredicate;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+class GenericTypeOfOperator implements Operator<Boolean> {
+    private final int symbol;
+    private final String displayName;
+    private final BiPredicate<Object, Object> operation;
+
+    public GenericTypeOfOperator(final int symbol, BiPredicate<Object, Object> operation) {
+        this.symbol = symbol;
+        displayName = DataPrepperExpressionParser.VOCABULARY.getDisplayName(symbol);
+        this.operation = operation;
+    }
+
+    @Override
+    public boolean shouldEvaluate(final RuleContext ctx) {
+        return ctx.getRuleIndex() == DataPrepperExpressionParser.RULE_typeOfOperatorExpression;
+    }
+
+    @Override
+    public int getSymbol() {
+        return symbol;
+    }
+
+    @Override
+    public Boolean evaluate(final Object ... args) {
+        checkArgument(args.length == 2, displayName + " requires operands length to be 2.");
+        return operation.test(args[0], args[1]);
+    }
+}

data-prepper-expression/src/main/java/org/opensearch/dataprepper/expression/OperatorConfiguration.java

@@ -6,6 +6,7 @@
 package org.opensearch.dataprepper.expression;
 
 import org.opensearch.dataprepper.expression.antlr.DataPrepperExpressionParser;
+import org.opensearch.dataprepper.model.event.DataType;
 import org.springframework.context.annotation.Bean;
 
 import javax.inject.Named;
@@ -22,6 +23,7 @@ class OperatorConfiguration {
     public final BiPredicate<Object, Object> regexEquals = (x, y) -> ((String) x).matches((String) y);
     public final BiPredicate<Object, Object> equals = Objects::equals;
     public final BiPredicate<Object, Object> inSet = (x, y) -> ((Set<?>) y).contains(x);
+    public final BiPredicate<Object, Object> typeOf = (x, y) -> DataType.isSameType(x, (String)y);
 
     @Bean
     public NumericCompareOperator greaterThanOperator() {
@@ -276,6 +278,11 @@ public AddBinaryOperator concatOperator() {
         return new AddBinaryOperator(DataPrepperExpressionParser.PLUS, null); 
     }
 
+    @Bean
+    public GenericTypeOfOperator typeofOperator() {
+        return new GenericTypeOfOperator(DataPrepperExpressionParser.TYPEOF, typeOf);
+    }
+
     @Bean
     public AddBinaryOperator addOperator() {
         final Map<Class<? extends Number>, Map<Class<? extends Number>, BiFunction<Object, Object, Number>>>