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;
  }
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