@Override
    public List<TrainingDataDTO> getTraningDataByCrisisAndAttribute(int crisisID,
                                                                    int modelFamilyID,
                                                                    int fromRecord,
                                                                    int limit,
                                                                    String sortColumn,
                                                                    String sortDirection) {
        List<TrainingDataDTO> trainingDataList = new ArrayList<TrainingDataDTO>();

        String orderSQLPart = "";
        if (sortColumn != null && !sortColumn.isEmpty()){
            if (sortDirection != null && !sortDirection.isEmpty()) {
                if ("ASC".equals(sortDirection)) {
                    sortDirection = "ASC";
                } else {
                    sortDirection = "DESC";
                }
            } else {
                sortDirection = "ASC";
            }
            orderSQLPart += " ORDER BY " + sortColumn + " " + sortDirection + " ";
        }

        String sql = " SELECT lbl.nominalLabelID, lbl.name labelName, d.data tweetJSON, u.userID, u.name labelerName, dnl.timestamp "
                + " FROM document_nominal_label dnl "
                + " JOIN nominal_label lbl on lbl.nominalLabelID=dnl.nominalLabelID "
                + " JOIN model_family mf on mf.nominalAttributeID=lbl.nominalAttributeID "
                //+ " JOIN model m on m.modelFamilyID= mf.modelFamilyID "
                + " JOIN document d on d.documentID = dnl.documentID "
                + " JOIN task_answer ta on ta.documentID = d.documentID "
                + " JOIN users u on u.userID = ta.userID "
                + " WHERE mf.modelFamilyID = :modelFamilyID AND d.crisisID = :crisisID " + orderSQLPart
                + " LIMIT :fromRecord, :limit";

        String sqlCount = " SELECT count(*) "
                + " FROM document_nominal_label dnl "
                + " JOIN nominal_label lbl on lbl.nominalLabelID=dnl.nominalLabelID "
                + " JOIN model_family mf on mf.nominalAttributeID=lbl.nominalAttributeID "
                //+ " JOIN model m on m.modelFamilyID= mf.modelFamilyID "
                + " JOIN document d on d.documentID = dnl.documentID "
                + " JOIN task_answer ta on ta.documentID = d.documentID "
                + " JOIN users u on u.userID = ta.userID "
                + " WHERE mf.modelFamilyID = :modelFamilyID AND d.crisisID = :crisisID ";
        try {
            Integer totalRows;

            Query queryCount = em.createNativeQuery(sqlCount);
            queryCount.setParameter("crisisID", crisisID);
            queryCount.setParameter("modelFamilyID", modelFamilyID);
            Object res = queryCount.getSingleResult();
            totalRows = Integer.parseInt(res.toString());

            if (totalRows > 0){
                Query query = em.createNativeQuery(sql);
                query.setParameter("crisisID", crisisID);
                query.setParameter("modelFamilyID", modelFamilyID);
                query.setParameter("fromRecord", fromRecord);
                query.setParameter("limit", limit);

                List<Object[]> rows = query.getResultList();
                TrainingDataDTO trainingDataRow;
                for (Object[] row : rows) {
                    trainingDataRow = new TrainingDataDTO();
                    trainingDataRow.setLabelID(((Integer) row[0]).intValue());
                    trainingDataRow.setLabelName((String) row[1]);
                    trainingDataRow.setTweetJSON((String) row[2]);
                    trainingDataRow.setLabelerID(((Integer) row[3]).intValue());
                    trainingDataRow.setLabelerName((String) row[4]);
                    trainingDataRow.setLabeledTime(((Date) row[5]));
                    trainingDataRow.setTotalRows(totalRows);
                    trainingDataList.add(trainingDataRow);
                }
            }
            return trainingDataList;
        } catch (NoResultException e) {
            return null;
        }
    }