@Nullable private Map<TopicPartition, KafkaInputSplit> parseAndOptimize(ExprNodeDesc expression) {
    if (expression.getClass() != ExprNodeGenericFuncDesc.class) {
      return null;
    }
    // get the kind of expression
    ExprNodeGenericFuncDesc expr = (ExprNodeGenericFuncDesc) expression;
    Class<?> op = expr.getGenericUDF().getClass();

    // handle the logical operators
    if (FunctionRegistry.isOpOr(expr)) {
      return pushOrOp(expr);
    }
    if (FunctionRegistry.isOpAnd(expr)) {
      return pushAndOp(expr);
    }

    if (op == GenericUDFOPGreaterThan.class) {
      return pushLeaf(expr, PredicateLeaf.Operator.LESS_THAN_EQUALS, true);
    } else if (op == GenericUDFOPEqualOrGreaterThan.class) {
      return pushLeaf(expr, PredicateLeaf.Operator.LESS_THAN, true);
    } else if (op == GenericUDFOPLessThan.class) {
      return pushLeaf(expr, PredicateLeaf.Operator.LESS_THAN, false);
    } else if (op == GenericUDFOPEqualOrLessThan.class) {
      return pushLeaf(expr, PredicateLeaf.Operator.LESS_THAN_EQUALS, false);
    } else if (op == GenericUDFOPEqual.class) {
      return pushLeaf(expr, PredicateLeaf.Operator.EQUALS, false);
      // otherwise, we didn't understand it, so bailout
    } else {
      return null;
    }
  }
@param expression filter to parseAndOptimize and trim the full scan

@return Map of optimized kafka range scans or null if it is impossible to optimize.
kafka-handler/src/java/org/apache/hadoop/hive/kafka/KafkaScanTrimmer.java