@VisibleForTesting static Map<TopicPartition, KafkaInputSplit> buildScanFromPartitionPredicate(Map<TopicPartition,
      KafkaInputSplit> fullScan,
      PredicateLeaf.Operator operator,
      int partitionConst,
      boolean flip,
      boolean negation) {
    final Predicate<TopicPartition> predicate;
    final Predicate<TopicPartition> intermediatePredicate;
    switch (operator) {
    case EQUALS:
      predicate = topicPartition -> topicPartition != null && topicPartition.partition() == partitionConst;
      break;
    case LESS_THAN:
      intermediatePredicate =
          flip ?
              topicPartition -> topicPartition != null && partitionConst < topicPartition.partition() :
              topicPartition -> topicPartition != null && topicPartition.partition() < partitionConst;

      predicate = negation ? intermediatePredicate.negate() : intermediatePredicate;
      break;
    case LESS_THAN_EQUALS:
      intermediatePredicate =
          flip ?
              topicPartition -> topicPartition != null && partitionConst <= topicPartition.partition() :
              topicPartition -> topicPartition != null && topicPartition.partition() <= partitionConst;

      predicate = negation ? intermediatePredicate.negate() : intermediatePredicate;
      break;
    default:
      //Default to select * for unknown cases
      predicate = topicPartition -> true;
    }

    ImmutableMap.Builder<TopicPartition, KafkaInputSplit> builder = ImmutableMap.builder();
    // Filter full scan based on predicate
    fullScan.entrySet()
        .stream()
        .filter(entry -> predicate.test(entry.getKey()))
        .forEach(entry -> builder.put(entry.getKey(), KafkaInputSplit.copyOf(entry.getValue())));
    return builder.build();
  }
Trim kafka scan using a leaf binary predicate on partition column.

@param fullScan       kafka full scan to be optimized
@param operator       predicate operator, equal, lessThan or lessThanEqual
@param partitionConst partition constant value
@param flip           true if the position of column and constant is flipped by default assuming column OP constant
@param negation       true if the expression is a negation of the original expression

@return filtered kafka scan
kafka-handler/src/java/org/apache/hadoop/hive/kafka/KafkaScanTrimmer.java