jdbc-protolib/src/main/java/com/streamsets/pipeline/lib/jdbc/JdbcMongoDBOplogRecordReader.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 = getFieldPath(columnName, columnsToFields, op);
      if (record.has(fieldPath)) {
        columnsToParameters.put(columnName, parameters.get(columnName));
      } else {
        LOG.trace("Record is missing a field for column {} for the operation code {}", columnName, op);
      }
    }
    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 = getFieldPath(columnName, columnsToFields, op);
      if (record.has(fieldPath)) {
        columnsToParameters.put(columnName, parameters.get(columnName));
      } else {
        LOG.trace("Record is missing a field for column {} for the operation code {}", columnName, op);
      }
    }
    return columnsToParameters;
  }
Records from MongoDB Oplog origin have a bit unique structure.

Records for Insert have all data in the field "o", which is a map and contains all data for columns
Records for Delete have objectId in the field "o", which is a map and contains only objectId.
Records for Update have a field "o2", which is a map and contains only objectId,
and a field "o" where it stores data for updating columns and values.

@param record
@param op
@param parameters
@param columnsToFields
@return