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;
  }
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
Split records according to tuples ( schemaName tableName )