private Map<TopicPartition, KafkaInputSplit> pushAndOp(ExprNodeGenericFuncDesc expr) {
    Map<TopicPartition, KafkaInputSplit> currentScan = new HashMap<>();

    fullHouse.forEach((tp, input) -> currentScan.put(tp, KafkaInputSplit.copyOf(input)));

    for (ExprNodeDesc child : expr.getChildren()) {
      Map<TopicPartition, KafkaInputSplit> scan = parseAndOptimize(child);
      if (scan != null) {
        Set<TopicPartition> currentKeys = ImmutableSet.copyOf(currentScan.keySet());
        currentKeys.forEach(key -> {
          KafkaInputSplit newSplit = scan.get(key);
          KafkaInputSplit oldSplit = currentScan.get(key);
          currentScan.remove(key);
          if (newSplit != null) {
            KafkaInputSplit intersectionSplit = KafkaInputSplit.intersectRange(newSplit, oldSplit);
            if (intersectionSplit != null) {
              currentScan.put(key, intersectionSplit);
            }
          }
        });

      }
    }
    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