@Nullable private Map<TopicPartition, KafkaInputSplit> pushOrOp(ExprNodeGenericFuncDesc expr) {
    final Map<TopicPartition, KafkaInputSplit> currentScan = new HashMap<>();
    for (ExprNodeDesc child : expr.getChildren()) {
      Map<TopicPartition, KafkaInputSplit> scan = parseAndOptimize(child);
      if (scan == null) {
        // if any of the children is unknown bailout
        return null;
      }

      scan.forEach((tp, input) -> {
        KafkaInputSplit existingSplit = currentScan.get(tp);
        currentScan.put(tp, KafkaInputSplit.unionRange(input, existingSplit == null ? input : existingSplit));
      });
    }
    return currentScan;
  }
@param expr And expression to be parsed

@return either full scan or an optimized sub scan.
kafka-handler/src/java/org/apache/hadoop/hive/kafka/KafkaScanTrimmer.java