/**
     * @param entityType
     * @param sourceIkey
     * @param targetIkey
     * @param sourceSchema
     * @param targetSchema
     * @param elementPrefix
     * @param compWhereClausure
     * @return
     * @throws Exception
     */
    protected ResultSet calculateAttributeDifferences(String entityType, String keyColumnName, String sourceIkey, String targetIkey,
                                                      String sourceSchema, String targetSchema, String elementPrefix,
                                                      String compWhereClausure, String additionalFilter
                                              ) throws Exception {

        String sql = null;
        ResultSet rs = null;

        //The list of columns is calculated yet
        if (!strColumnsWithToCharMap.containsKey(elementPrefix)) {
            getListOfcolumnsForSelect(elementPrefix);
        }

        String twoValue="targetValue";
        String oneValue="sourceValue";

        if (targetSchema.endsWith("_W.")) {
            twoValue="sourceValue";
            oneValue="targetValue";
        }


        sql = "select col, sourceValue, targetValue" +
        " from (" +
        " select col," +
        "   max(case when rn = 2 then val end) as " + twoValue + "," +
        "   max(case when rn = 1 then val end) as " + oneValue +
        " from (" +
        "    select u.*, row_number() over (partition by u.col order by u." + keyColumnName +") as rn" +
        "  from (" +
        "  select " + keyColumnName+ ", " + getStrColumnsWithToChar(elementPrefix) +
        "    from " + sourceSchema + entityType + " WHERE " + keyColumnName + "=" + sourceIkey + compWhereClausure +
        "    UNION" +
        "    select " + keyColumnName+ ", " + getStrColumnsWithToChar(elementPrefix) +
        "    from " + targetSchema + entityType +" WHERE " + keyColumnName + "=" + targetIkey + compWhereClausure +
        "    order by " + keyColumnName +
        "   ) unpivot INCLUDE NULLS (val for col in (" + getStrColumns(elementPrefix) + ")) u" +
        " )" +
        " group by col" +
        " ) where (sourceValue<>targetValue OR (sourceValue is NULL and targetValue is not NULL)) ";
        //" OR (sourceValue is not NULL and targetValue is NULL)) ";
        if (!TextUtil.isEmpty(additionalFilter)) {
            sql += additionalFilter;
        }

        try {

            // Connect and query to the database
            rs = StatementManager.executePreparedQuery(conn, sql);


        } catch (SQLException e) {
            throw new SRDBSQLException(e, sql);
        } catch (Exception ex) {
            throw new SRDBException(ex);
        }

        return rs;


    }