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);
  }
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.
Build query using the lastOffset which is of the form ( <column1 > = <value1 > :: <column2 > = <value2 > :: <column3 > = <value3 > )