@Nullable private Map<TopicPartition, KafkaInputSplit> pushLeaf(ExprNodeGenericFuncDesc expr,
      PredicateLeaf.Operator operator,
      boolean negation) {
    if (expr.getChildren().size() != 2) {
      return null;
    }
    GenericUDF genericUDF = expr.getGenericUDF();
    if (!(genericUDF instanceof GenericUDFBaseCompare)) {
      return null;
    }
    ExprNodeDesc expr1 = expr.getChildren().get(0);
    ExprNodeDesc expr2 = expr.getChildren().get(1);
    // We may need to peel off the GenericUDFBridge that is added by CBO or user
    if (expr1.getTypeInfo().equals(expr2.getTypeInfo())) {
      expr1 = getColumnExpr(expr1);
      expr2 = getColumnExpr(expr2);
    }

    ExprNodeDesc[] extracted = ExprNodeDescUtils.extractComparePair(expr1, expr2);
    if (extracted == null || (extracted.length > 2)) {
      return null;
    }

    ExprNodeColumnDesc columnDesc;
    ExprNodeConstantDesc constantDesc;
    final boolean flip;

    if (extracted[0] instanceof ExprNodeColumnDesc) {
      columnDesc = (ExprNodeColumnDesc) extracted[0];
      constantDesc = (ExprNodeConstantDesc) extracted[1];
      flip = false;

    } else {
      flip = true;
      columnDesc = (ExprNodeColumnDesc) extracted[1];
      constantDesc = (ExprNodeConstantDesc) extracted[0];
    }

    if (columnDesc.getColumn().equals(MetadataColumn.PARTITION.getName())) {
      return buildScanFromPartitionPredicate(fullHouse,
          operator,
          ((Number) constantDesc.getValue()).intValue(),
          flip,
          negation);

    }
    if (columnDesc.getColumn().equals(MetadataColumn.OFFSET.getName())) {
      return buildScanFromOffsetPredicate(fullHouse,
          operator,
          ((Number) constantDesc.getValue()).longValue(),
          flip,
          negation);
    }

    if (columnDesc.getColumn().equals(MetadataColumn.TIMESTAMP.getName())) {
      long timestamp = ((Number) constantDesc.getValue()).longValue();
      //noinspection unchecked
      return buildScanForTimesPredicate(fullHouse, operator, timestamp, flip, negation, kafkaConsumer);
    }
    return null;
  }
@param expr     leaf node to push
@param operator operator
@param negation true if it is a negation, this is used to represent:
                GenericUDFOPGreaterThan and GenericUDFOPEqualOrGreaterThan
                using PredicateLeaf.Operator.LESS_THAN and PredicateLeaf.Operator.LESS_THAN_EQUALS

@return leaf scan or null if can not figure out push down
kafka-handler/src/java/org/apache/hadoop/hive/kafka/KafkaScanTrimmer.java