@VisibleForTesting static Map<TopicPartition, KafkaInputSplit> buildScanFromOffsetPredicate(Map<TopicPartition,
      KafkaInputSplit> fullScan,
      PredicateLeaf.Operator operator,
      long offsetConst,
      boolean flip,
      boolean negation) {
    final boolean isEndBound;
    final long startOffset;
    final long endOffset;

    isEndBound = flip == negation;
    switch (operator) {
    case LESS_THAN_EQUALS:
      if (isEndBound) {
        startOffset = -1;
        endOffset = negation ? offsetConst : offsetConst + 1;
      } else {
        endOffset = -1;
        startOffset = negation ? offsetConst + 1 : offsetConst;
      }
      break;
    case EQUALS:
      startOffset = offsetConst;
      endOffset = offsetConst + 1;
      break;
    case LESS_THAN:
      if (isEndBound) {
        endOffset = negation ? offsetConst + 1 : offsetConst;
        startOffset = -1;
      } else {
        endOffset = -1;
        startOffset = negation ? offsetConst : offsetConst + 1;
      }
      break;
    default:
      // default to select *
      startOffset = -1;
      endOffset = -1;
    }

    final Map<TopicPartition, KafkaInputSplit> newScan = new HashMap<>();

    fullScan.forEach((tp, existingInputSplit) -> {
      final KafkaInputSplit newInputSplit;
      if (startOffset != -1 && endOffset == -1) {
        newInputSplit = new KafkaInputSplit(tp.topic(),
            tp.partition(),
            //if the user ask for start offset > max offset will replace with last offset
            Math.min(startOffset, existingInputSplit.getEndOffset()),
            existingInputSplit.getEndOffset(),
            existingInputSplit.getPath());
      } else if (endOffset != -1 && startOffset == -1) {
        newInputSplit = new KafkaInputSplit(tp.topic(), tp.partition(), existingInputSplit.getStartOffset(),
            //@TODO check this, if user ask for non existing end offset ignore it and position head on start
            // This can be an issue when doing ingestion from kafka into Hive, what happen if there is some gaps
            // Shall we fail the ingest or carry-on and ignore non existing offsets
            Math.max(endOffset, existingInputSplit.getStartOffset()), existingInputSplit.getPath());
      } else if (endOffset == startOffset + 1) {
        if (startOffset < existingInputSplit.getStartOffset() || startOffset >= existingInputSplit.getEndOffset()) {
          newInputSplit = new KafkaInputSplit(tp.topic(), tp.partition(),
              // non existing offset will be seeking last offset
              existingInputSplit.getEndOffset(), existingInputSplit.getEndOffset(), existingInputSplit.getPath());
        } else {
          newInputSplit =
              new KafkaInputSplit(tp.topic(), tp.partition(), startOffset, endOffset, existingInputSplit.getPath());
        }

      } else {
        newInputSplit =
            new KafkaInputSplit(tp.topic(),
                tp.partition(),
                existingInputSplit.getStartOffset(),
                existingInputSplit.getEndOffset(),
                existingInputSplit.getPath());
      }

      newScan.put(tp, KafkaInputSplit.intersectRange(newInputSplit, existingInputSplit));
    });

    return newScan;
  }
@param fullScan    full kafka scan to be pruned
@param operator    operator kind
@param offsetConst offset constant value
@param flip        true if position of constant and column were flipped by default assuming COLUMN OP CONSTANT
@param negation    true if the expression is a negation of the original expression

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