jdbc-protolib/src/main/java/com/streamsets/pipeline/lib/jdbc/JdbcMongoDBOplogRecordReader.java
@Override
  String getFieldPath(String columnName, Map<String, String> columnsToField, int op) {
    if (op == OperationType.UPDATE_CODE){
      String fieldPath = columnsToField.get(columnName);
      if (fieldPath == null){
        LOG.error("Column name {} is not defined in column-filed mapping", columnName);
        return null;
      }

      if (fieldPath.contains(ID_FIELD)) {
        // _id is stored in "/o2/column_name for update records. Need to change the fieldpath"
        return fieldPath.replace(OP_FIELD, OP2_FIELD);
      } else {
        // column and values are stored in "/o/$set/column_name". Need to change the fieldpath
        return fieldPath.replaceFirst(OP_FIELD, String.format("%s/%s", OP_FIELD, SET_FIELD));
      }

    }
    return columnsToField.get(columnName);
  }
@Override
  String getFieldPath(String columnName, Map<String, String> columnsToField, int op) {
    if (op == OperationType.UPDATE_CODE){
      String fieldPath = columnsToField.get(columnName);
      if (fieldPath == null){
        LOG.error("Column name {} is not defined in column-filed mapping", columnName);
        return null;
      }

      if (fieldPath.contains(ID_FIELD)) {
        // _id is stored in "/o2/column_name for update records. Need to change the fieldpath"
        return fieldPath.replace(OP_FIELD, OP2_FIELD);
      } else {
        // column and values are stored in "/o/$set/column_name". Need to change the fieldpath
        return fieldPath.replaceFirst(OP_FIELD, String.format("%s/%s", OP_FIELD, SET_FIELD));
      }

    }
    return columnsToField.get(columnName);
  }
@ Override String getFieldPath ( String columnName , Map < String , String > columnsToField , int op ) { if ( op == OperationType . UPDATE_CODE ) { String fieldPath = columnsToField . get ( columnName ) ; if ( fieldPath == null ) { LOG . error ( "Column name {} is not defined in column-filed mapping" , columnName ) ; return null ; } if ( fieldPath . contains ( ID_FIELD ) ) { // _id is stored in "/o2/column_name for update records. Need to change the fieldpath" return fieldPath . replace ( OP_FIELD , OP2_FIELD ) ; } else { // column and values are stored in "/o/$set/column_name". Need to change the fieldpath return fieldPath . replaceFirst ( OP_FIELD , String . format ( "%s/%s" , OP_FIELD , SET_FIELD ) ) ; } } return columnsToField . get ( columnName ) ; }
Return a fieldpath for the corresponding column name. The columnToField Map
contains column-to-field mapping, so simply get fieldpath for the column name
if operation is INSERT or DELETE.
We replace /o with /o2 to get _id field, and change /o to /o/$set to get
updating column & value fields.
@param columnName column name
@param columnsToField mapping of column name to field path
@param op  Not used here but is used in subclass.
@return
Return a fieldpath for the corresponding column name . The columnToField Map contains column - to - field mapping so simply get fieldpath for the column name if operation is INSERT or DELETE . We replace / o with / o2 to get _id field and change / o to / o / $set to get updating column & value fields .