jdbc-protolib/src/main/java/com/streamsets/pipeline/lib/jdbc/SchemaTableClassifier.java
public Multimap<SchemaAndTable, Record> classify(Batch batch) throws OnRecordErrorException {
    Multimap<SchemaAndTable, Record> partitions = ArrayListMultimap.create();
    Iterator<Record> batchIterator = batch.getRecords();

    while (batchIterator.hasNext()) {
      Record record = batchIterator.next();
      String schemaName = schemaNameExpr;
      String tableName = tableNameExpr;

      if (dynamicSchemaName) {
        try {
          RecordEL.setRecordInContext(schemaNameVars, record);
          schemaName = schemaNameEval.eval(schemaNameVars, schemaNameExpr, String.class);
          LOG.debug("Expression '{}' is evaluated to '{}' : ", schemaNameExpr, schemaName);
        } catch (ELEvalException e) {
          LOG.error("Failed to evaluate expression '{}' : ", schemaNameExpr, e.toString(), e);
          throw new OnRecordErrorException(record, e.getErrorCode(), e.getParams());
        }
      }

      if (dynamicTableName) {
        try {
          RecordEL.setRecordInContext(tableNameVars, record);
          tableName = tableNameEval.eval(tableNameVars, tableNameExpr, String.class);
          LOG.debug("Expression '{}' is evaluated to '{}' : ", tableNameExpr, tableName);
        } catch (ELEvalException e) {
          LOG.error("Failed to evaluate expression '{}' : ", tableNameExpr, e.toString(), e);
          throw new OnRecordErrorException(record, e.getErrorCode(), e.getParams());
        }
      }

      SchemaAndTable partitionKey = new SchemaAndTable(schemaName, tableName);
      partitions.put(partitionKey, record);
    }

    return partitions;
  }
public Multimap<SchemaAndTable, Record> classify(Batch batch) throws OnRecordErrorException {
    Multimap<SchemaAndTable, Record> partitions = ArrayListMultimap.create();
    Iterator<Record> batchIterator = batch.getRecords();

    while (batchIterator.hasNext()) {
      Record record = batchIterator.next();
      String schemaName = schemaNameExpr;
      String tableName = tableNameExpr;

      if (dynamicSchemaName) {
        try {
          RecordEL.setRecordInContext(schemaNameVars, record);
          schemaName = schemaNameEval.eval(schemaNameVars, schemaNameExpr, String.class);
          LOG.debug("Expression '{}' is evaluated to '{}' : ", schemaNameExpr, schemaName);
        } catch (ELEvalException e) {
          LOG.error("Failed to evaluate expression '{}' : ", schemaNameExpr, e.toString(), e);
          throw new OnRecordErrorException(record, e.getErrorCode(), e.getParams());
        }
      }

      if (dynamicTableName) {
        try {
          RecordEL.setRecordInContext(tableNameVars, record);
          tableName = tableNameEval.eval(tableNameVars, tableNameExpr, String.class);
          LOG.debug("Expression '{}' is evaluated to '{}' : ", tableNameExpr, tableName);
        } catch (ELEvalException e) {
          LOG.error("Failed to evaluate expression '{}' : ", tableNameExpr, e.toString(), e);
          throw new OnRecordErrorException(record, e.getErrorCode(), e.getParams());
        }
      }

      SchemaAndTable partitionKey = new SchemaAndTable(schemaName, tableName);
      partitions.put(partitionKey, record);
    }

    return partitions;
  }
Split records according to tuples (schemaName, tableName)

@param batch batch of SDC records
@return batch partitions grouped by the tuples (schemaName, tableName)
@throws OnRecordErrorException