jdbc-protolib/src/main/java/com/streamsets/pipeline/lib/jdbc/multithread/util/OffsetQueryUtil.java
public static Pair<String, List<Pair<Integer, String>>> buildAndReturnQueryAndParamValToSet(
      TableRuntimeContext tableRuntimeContext,
      String lastOffset,
      String quoteChar,
      TableJdbcELEvalContext tableJdbcELEvalContext
  ) throws ELEvalException {
    final TableContext tableContext = tableRuntimeContext.getSourceTableContext();
    StringBuilder queryBuilder = new StringBuilder();
    List<Pair<Integer, String>> paramValueToSet = new ArrayList<>();
    queryBuilder.append(buildBaseTableQuery(tableRuntimeContext, quoteChar));

    Map<String, String> storedTableToOffset = getColumnsToOffsetMapFromOffsetFormat(lastOffset);
    final boolean noStoredOffsets = storedTableToOffset.isEmpty();

    //Determines whether an initial offset is specified in the config and there is no stored offset.
    boolean isOffsetOverriden = tableContext.isOffsetOverriden() && noStoredOffsets;

    OffsetComparison minComparison = OffsetComparison.GREATER_THAN;
    Map<String, String> offset = null;
    if (isOffsetOverriden) {
      //Use the offset in the configuration
      offset = tableContext.getOffsetColumnToStartOffset();
    } else if (tableRuntimeContext.isPartitioned() && noStoredOffsets) {
      // use partitioned starting offsets
      offset = tableRuntimeContext.getStartingPartitionOffsets();
      minComparison = OffsetComparison.GREATER_THAN_OR_EQUALS;
    } else {
      // if offset is available
      // get the stored offset (which is of the form partitionName=value) and strip off 'offsetColumns=' prefix
      // else null
      // offset = storedOffsets;
      offset = storedTableToOffset;
    }

    Map<String, String> maxOffsets = new HashMap<>();
    if (tableRuntimeContext.isPartitioned()) {
      maxOffsets.putAll(tableRuntimeContext.getMaxPartitionOffsets());
    }

    List<String> finalAndConditions = new ArrayList<>();
    //Apply last offset conditions
    if (offset != null && !offset.isEmpty()) {
      List<String> finalOrConditions = new ArrayList<>();
      List<String> preconditions = new ArrayList<>();
      List<Pair<Integer, String>> preconditionParamVals = new ArrayList<>();
      //For partition columns p1, p2 and p3 with offsets o1, o2 and o3 respectively, the query will look something like
      //select * from tableName where (p1 > o1) or (p1 = o1 and p2 > o2) or (p1 = o1 and p2 = o2 and p3 > o3) order by p1, p2, p3.
      for (String partitionColumn : tableContext.getOffsetColumns()) {
        int partitionSqlType = tableContext.getOffsetColumnType(partitionColumn);
        String partitionOffset = offset.get(partitionColumn);


        String thisPartitionColumnMax = null;
        boolean hasMaxOffset = false;
        // add max value for partition column (if applicable)
        if (maxOffsets.containsKey(partitionColumn)) {
          final String maxOffset = maxOffsets.get(partitionColumn);
          if (!Strings.isNullOrEmpty(maxOffset)) {
            Pair<Integer, String> paramValForCurrentOffsetColumnMax = Pair.of(partitionSqlType, maxOffset);
            // add for current partition column max value
            paramValueToSet.add(paramValForCurrentOffsetColumnMax);
            thisPartitionColumnMax =
                getConditionForPartitionColumn(
                    partitionColumn,
                    OffsetComparison.LESS_THAN,
                    preconditions,
                    quoteChar
                );
            hasMaxOffset = true;
          }
        }

        final String thisPartitionColumnMin = getConditionForPartitionColumn(partitionColumn,
            minComparison,
            preconditions,
            quoteChar
        );

        String conditionForThisPartitionColumn;
        if (hasMaxOffset) {
          conditionForThisPartitionColumn = String.format(
              AND_CONDITION_FORMAT,
              // add max condition first, since its param to set was already added
              thisPartitionColumnMax,
              thisPartitionColumnMin
          );
        } else {
          conditionForThisPartitionColumn = String.format(CONDITION_FORMAT, thisPartitionColumnMin);
        }

        //Add for preconditions (EX: composite keys)
        paramValueToSet.addAll(new ArrayList<>(preconditionParamVals));
        Pair<Integer, String> paramValForCurrentOffsetColumn = Pair.of(partitionSqlType, partitionOffset);
        //Add for current partition column
        paramValueToSet.add(paramValForCurrentOffsetColumn);
        finalOrConditions.add(conditionForThisPartitionColumn);
        preconditions.add(
            getConditionForPartitionColumn(
                partitionColumn,
                OffsetComparison.EQUALS,
                Collections.emptyList(),
                quoteChar
            )
        );

        preconditionParamVals.add(paramValForCurrentOffsetColumn);
      }
      finalAndConditions.add(String.format(CONDITION_FORMAT, OR_JOINER.join(finalOrConditions)));
    }

    if (!StringUtils.isEmpty(tableContext.getExtraOffsetColumnConditions())) {
      //Apply extra offset column conditions configured which will be appended as AND on the query
      String condition =
          tableJdbcELEvalContext.evaluateAsString(
              "extraOffsetColumnConditions",
              tableContext.getExtraOffsetColumnConditions()
          );
      finalAndConditions.add(String.format(CONDITION_FORMAT, condition));
    }

    if (!finalAndConditions.isEmpty()) {
      queryBuilder.append(String.format(WHERE_CLAUSE, AND_JOINER.join(finalAndConditions)));
    }

    Collection<String> quotedOffsetColumns =
        tableContext.getOffsetColumns().stream().map(
            offsetCol -> String.format(QUOTED_NAME, quoteChar, offsetCol, quoteChar)
        ).collect(Collectors.toList());

    queryBuilder.append(String.format(ORDER_BY_CLAUSE, COMMA_SPACE_JOINER.join(quotedOffsetColumns)));
    return Pair.of(queryBuilder.toString(),paramValueToSet);
  }
public static Pair<String, List<Pair<Integer, String>>> buildAndReturnQueryAndParamValToSet(
      TableRuntimeContext tableRuntimeContext,
      String lastOffset,
      String quoteChar,
      TableJdbcELEvalContext tableJdbcELEvalContext
  ) throws ELEvalException {
    final TableContext tableContext = tableRuntimeContext.getSourceTableContext();
    StringBuilder queryBuilder = new StringBuilder();
    List<Pair<Integer, String>> paramValueToSet = new ArrayList<>();
    queryBuilder.append(buildBaseTableQuery(tableRuntimeContext, quoteChar));

    Map<String, String> storedTableToOffset = getColumnsToOffsetMapFromOffsetFormat(lastOffset);
    final boolean noStoredOffsets = storedTableToOffset.isEmpty();

    //Determines whether an initial offset is specified in the config and there is no stored offset.
    boolean isOffsetOverriden = tableContext.isOffsetOverriden() && noStoredOffsets;

    OffsetComparison minComparison = OffsetComparison.GREATER_THAN;
    Map<String, String> offset = null;
    if (isOffsetOverriden) {
      //Use the offset in the configuration
      offset = tableContext.getOffsetColumnToStartOffset();
    } else if (tableRuntimeContext.isPartitioned() && noStoredOffsets) {
      // use partitioned starting offsets
      offset = tableRuntimeContext.getStartingPartitionOffsets();
      minComparison = OffsetComparison.GREATER_THAN_OR_EQUALS;
    } else {
      // if offset is available
      // get the stored offset (which is of the form partitionName=value) and strip off 'offsetColumns=' prefix
      // else null
      // offset = storedOffsets;
      offset = storedTableToOffset;
    }

    Map<String, String> maxOffsets = new HashMap<>();
    if (tableRuntimeContext.isPartitioned()) {
      maxOffsets.putAll(tableRuntimeContext.getMaxPartitionOffsets());
    }

    List<String> finalAndConditions = new ArrayList<>();
    //Apply last offset conditions
    if (offset != null && !offset.isEmpty()) {
      List<String> finalOrConditions = new ArrayList<>();
      List<String> preconditions = new ArrayList<>();
      List<Pair<Integer, String>> preconditionParamVals = new ArrayList<>();
      //For partition columns p1, p2 and p3 with offsets o1, o2 and o3 respectively, the query will look something like
      //select * from tableName where (p1 > o1) or (p1 = o1 and p2 > o2) or (p1 = o1 and p2 = o2 and p3 > o3) order by p1, p2, p3.
      for (String partitionColumn : tableContext.getOffsetColumns()) {
        int partitionSqlType = tableContext.getOffsetColumnType(partitionColumn);
        String partitionOffset = offset.get(partitionColumn);


        String thisPartitionColumnMax = null;
        boolean hasMaxOffset = false;
        // add max value for partition column (if applicable)
        if (maxOffsets.containsKey(partitionColumn)) {
          final String maxOffset = maxOffsets.get(partitionColumn);
          if (!Strings.isNullOrEmpty(maxOffset)) {
            Pair<Integer, String> paramValForCurrentOffsetColumnMax = Pair.of(partitionSqlType, maxOffset);
            // add for current partition column max value
            paramValueToSet.add(paramValForCurrentOffsetColumnMax);
            thisPartitionColumnMax =
                getConditionForPartitionColumn(
                    partitionColumn,
                    OffsetComparison.LESS_THAN,
                    preconditions,
                    quoteChar
                );
            hasMaxOffset = true;
          }
        }

        final String thisPartitionColumnMin = getConditionForPartitionColumn(partitionColumn,
            minComparison,
            preconditions,
            quoteChar
        );

        String conditionForThisPartitionColumn;
        if (hasMaxOffset) {
          conditionForThisPartitionColumn = String.format(
              AND_CONDITION_FORMAT,
              // add max condition first, since its param to set was already added
              thisPartitionColumnMax,
              thisPartitionColumnMin
          );
        } else {
          conditionForThisPartitionColumn = String.format(CONDITION_FORMAT, thisPartitionColumnMin);
        }

        //Add for preconditions (EX: composite keys)
        paramValueToSet.addAll(new ArrayList<>(preconditionParamVals));
        Pair<Integer, String> paramValForCurrentOffsetColumn = Pair.of(partitionSqlType, partitionOffset);
        //Add for current partition column
        paramValueToSet.add(paramValForCurrentOffsetColumn);
        finalOrConditions.add(conditionForThisPartitionColumn);
        preconditions.add(
            getConditionForPartitionColumn(
                partitionColumn,
                OffsetComparison.EQUALS,
                Collections.emptyList(),
                quoteChar
            )
        );

        preconditionParamVals.add(paramValForCurrentOffsetColumn);
      }
      finalAndConditions.add(String.format(CONDITION_FORMAT, OR_JOINER.join(finalOrConditions)));
    }

    if (!StringUtils.isEmpty(tableContext.getExtraOffsetColumnConditions())) {
      //Apply extra offset column conditions configured which will be appended as AND on the query
      String condition =
          tableJdbcELEvalContext.evaluateAsString(
              "extraOffsetColumnConditions",
              tableContext.getExtraOffsetColumnConditions()
          );
      finalAndConditions.add(String.format(CONDITION_FORMAT, condition));
    }

    if (!finalAndConditions.isEmpty()) {
      queryBuilder.append(String.format(WHERE_CLAUSE, AND_JOINER.join(finalAndConditions)));
    }

    Collection<String> quotedOffsetColumns =
        tableContext.getOffsetColumns().stream().map(
            offsetCol -> String.format(QUOTED_NAME, quoteChar, offsetCol, quoteChar)
        ).collect(Collectors.toList());

    queryBuilder.append(String.format(ORDER_BY_CLAUSE, COMMA_SPACE_JOINER.join(quotedOffsetColumns)));
    return Pair.of(queryBuilder.toString(),paramValueToSet);
  }
Build query using the lastOffset which is of the form (<column1>=<value1>::<column2>=<value2>::<column3>=<value3>)

@param tableRuntimeContext Context for the current table.
@param lastOffset the last offset for this particular table
@return A query to execute for the current batch.