jdbc-protolib/src/main/java/com/streamsets/pipeline/lib/jdbc/JdbcMySqlBinLogRecordReader.java
@Override
  public SortedMap<String, String> getColumnsToParameters(
      final Record record, int op,
      Map<String, String> parameters,
      Map<String, String> columnsToFields)
  {
    SortedMap<String, String> columnsToParameters = new TreeMap<>();
    for (Map.Entry<String, String> entry : columnsToFields.entrySet()) {
      String columnName = entry.getKey();
      String fieldPath = entry.getValue();
      if(op == OperationType.DELETE_CODE){
        fieldPath = fieldPath.replace(DATA_FIELD, OLD_DATA_FIELD);
      }

      if (record.has(fieldPath)) {
        columnsToParameters.put(columnName, parameters.get(columnName));
      }
    }
    return columnsToParameters;
  }
@Override
  public SortedMap<String, String> getColumnsToParameters(
      final Record record, int op,
      Map<String, String> parameters,
      Map<String, String> columnsToFields)
  {
    SortedMap<String, String> columnsToParameters = new TreeMap<>();
    for (Map.Entry<String, String> entry : columnsToFields.entrySet()) {
      String columnName = entry.getKey();
      String fieldPath = entry.getValue();
      if(op == OperationType.DELETE_CODE){
        fieldPath = fieldPath.replace(DATA_FIELD, OLD_DATA_FIELD);
      }

      if (record.has(fieldPath)) {
        columnsToParameters.put(columnName, parameters.get(columnName));
      }
    }
    return columnsToParameters;
  }
@ Override public SortedMap < String , String > getColumnsToParameters ( final Record record , int op , Map < String , String > parameters , Map < String , String > columnsToFields ) { SortedMap < String , String > columnsToParameters = new TreeMap <> ( ) ; for ( Map . Entry < String , String > entry : columnsToFields . entrySet ( ) ) { String columnName = entry . getKey ( ) ; String fieldPath = entry . getValue ( ) ; if ( op == OperationType . DELETE_CODE ) { fieldPath = fieldPath . replace ( DATA_FIELD , OLD_DATA_FIELD ) ; } if ( record . has ( fieldPath ) ) { columnsToParameters . put ( columnName , parameters . get ( columnName ) ) ; } } return columnsToParameters ; }
Records from MySQL BinLog origin have a bit unique structure.

Records for Insert and update operations have a field path "/Data", which is a map
of all column names as key and values stored in DB as value.

Record for Delete operation don't have /Data field. Instead it has a filed path "/OldData",
which store the original data in table. So need to look into /OldData when operation is delete.

We expect user to configure /Data as a filed path in column-filed mapping configuration.
@param record
@param op
@param parameters
@param columnsToFields
@return
Records from MySQL BinLog origin have a bit unique structure .