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;
  }
@ 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
Records from MongoDB Oplog origin have a bit unique structure .