public static QueryInfo generateQueryInfo(JavaSparkContext jssc, String srcBasePath,
                                            int numInstantsPerFetch, Option<String> beginInstant,
                                            MissingCheckpointStrategy missingCheckpointStrategy,
                                            HollowCommitHandling handlingMode,
                                            String orderColumn, String keyColumn, String limitColumn,
                                            boolean sourceLimitBasedBatching,
                                            Option<String> lastCheckpointKey) {
    ValidationUtils.checkArgument(numInstantsPerFetch > 0,
        "Make sure the config hoodie.streamer.source.hoodieincr.num_instants is set to a positive value");
    HoodieTableMetaClient srcMetaClient = HoodieTableMetaClient.builder()
        .setConf(HadoopFSUtils.getStorageConfWithCopy(jssc.hadoopConfiguration()))
        .setBasePath(srcBasePath).setLoadActiveTimelineOnLoad(true).build();

    HoodieTimeline completedCommitTimeline = srcMetaClient.getCommitsAndCompactionTimeline().filterCompletedInstants();
    final HoodieTimeline activeCommitTimeline = handleHollowCommitIfNeeded(completedCommitTimeline, srcMetaClient, handlingMode);
    Function<HoodieInstant, String> timestampForLastInstant = instant -> handlingMode == HollowCommitHandling.USE_TRANSITION_TIME
        ? instant.getCompletionTime() : instant.getTimestamp();
    String beginInstantTime = beginInstant.orElseGet(() -> {
      if (missingCheckpointStrategy != null) {
        if (missingCheckpointStrategy == MissingCheckpointStrategy.READ_LATEST) {
          Option<HoodieInstant> lastInstant = activeCommitTimeline.lastInstant();
          return lastInstant.map(hoodieInstant -> getStrictlyLowerTimestamp(timestampForLastInstant.apply(hoodieInstant))).orElse(DEFAULT_BEGIN_TIMESTAMP);
        } else {
          return DEFAULT_BEGIN_TIMESTAMP;
        }
      } else {
        throw new IllegalArgumentException("Missing begin instant for incremental pull. For reading from latest "
            + "committed instant set hoodie.streamer.source.hoodieincr.missing.checkpoint.strategy to a valid value");
      }
    });

    // When `beginInstantTime` is present, `previousInstantTime` is set to the completed commit before `beginInstantTime` if that exists.
    // If there is no completed commit before `beginInstantTime`, e.g., `beginInstantTime` is the first commit in the active timeline,
    // `previousInstantTime` is set to `DEFAULT_BEGIN_TIMESTAMP`.
    String previousInstantTime = DEFAULT_BEGIN_TIMESTAMP;
    if (!beginInstantTime.equals(DEFAULT_BEGIN_TIMESTAMP)) {
      Option<HoodieInstant> previousInstant = activeCommitTimeline.findInstantBefore(beginInstantTime);
      if (previousInstant.isPresent()) {
        previousInstantTime = previousInstant.get().getTimestamp();
      } else {
        // if begin instant time matches first entry in active timeline, we can set previous = beginInstantTime - 1
        if (activeCommitTimeline.filterCompletedInstants().firstInstant().isPresent()
            && activeCommitTimeline.filterCompletedInstants().firstInstant().get().getTimestamp().equals(beginInstantTime)) {
          previousInstantTime = String.valueOf(Long.parseLong(beginInstantTime) - 1);
        }
      }
    }

    if (missingCheckpointStrategy == MissingCheckpointStrategy.READ_LATEST || !activeCommitTimeline.isBeforeTimelineStarts(beginInstantTime)) {
      Option<HoodieInstant> nthInstant;
      // When we are in the upgrade code path from non-sourcelimit-based batching to sourcelimit-based batching, we need to avoid fetching the commit
      // that is read already. Else we will have duplicates in append-only use case if we use "findInstantsAfterOrEquals".
      // As soon as we have a new format of checkpoint and a key we will move to the new code of fetching the current commit as well.
      if (sourceLimitBasedBatching && lastCheckpointKey.isPresent()) {
        nthInstant = Option.fromJavaOptional(activeCommitTimeline
            .findInstantsAfterOrEquals(beginInstantTime, numInstantsPerFetch).getInstantsAsStream().reduce((x, y) -> y));
      } else {
        nthInstant = Option.fromJavaOptional(activeCommitTimeline
            .findInstantsAfter(beginInstantTime, numInstantsPerFetch).getInstantsAsStream().reduce((x, y) -> y));
      }
      return new QueryInfo(DataSourceReadOptions.QUERY_TYPE_INCREMENTAL_OPT_VAL(), previousInstantTime,
          beginInstantTime, nthInstant.map(HoodieInstant::getTimestamp).orElse(beginInstantTime),
          orderColumn, keyColumn, limitColumn);
    } else {
      // when MissingCheckpointStrategy is set to read everything until latest, trigger snapshot query.
      Option<HoodieInstant> lastInstant = activeCommitTimeline.lastInstant();
      return new QueryInfo(DataSourceReadOptions.QUERY_TYPE_SNAPSHOT_OPT_VAL(),
          previousInstantTime, beginInstantTime, lastInstant.get().getTimestamp(),
          orderColumn, keyColumn, limitColumn);
    }
  }
@Test
  void testQueryInfoGeneration() throws IOException {
    String commitTimeForReads = "1";
    String commitTimeForWrites = "2";

    Pair<String, List<HoodieRecord>> inserts = writeS3MetadataRecords(commitTimeForReads);
    inserts = writeS3MetadataRecords(commitTimeForWrites);

    String startInstant = commitTimeForReads;
    String orderColumn = "_hoodie_commit_time";
    String keyColumn = "s3.object.key";
    String limitColumn = "s3.object.size";
    QueryInfo queryInfo = IncrSourceHelper.generateQueryInfo(jsc, basePath(), 5, Option.of(startInstant), null,
        TimelineUtils.HollowCommitHandling.BLOCK, orderColumn, keyColumn, limitColumn, true, Option.empty());
    assertEquals(String.valueOf(Integer.parseInt(commitTimeForReads) - 1), queryInfo.getPreviousInstant());
    assertEquals(commitTimeForReads, queryInfo.getStartInstant());
    assertEquals(commitTimeForWrites, queryInfo.getEndInstant());

    startInstant = commitTimeForWrites;
    queryInfo = IncrSourceHelper.generateQueryInfo(jsc, basePath(), 5, Option.of(startInstant), null,
        TimelineUtils.HollowCommitHandling.BLOCK, orderColumn, keyColumn, limitColumn, true, Option.empty());
    assertEquals(commitTimeForReads, queryInfo.getPreviousInstant());
    assertEquals(commitTimeForWrites, queryInfo.getStartInstant());
    assertEquals(commitTimeForWrites, queryInfo.getEndInstant());

  }