datacleaning.java

数据仓库工具

    String addins="";
    String prefix="";
    String oidVersion="";
    if(oid){
//      oidVersion=", version="+currentVersion;
      oidVersion=", "+versionColumnName+"="+currentVersion;
    }
    if(typeOfInsert.equalsIgnoreCase("update")){
      addins=oidVersion+" where ";
      prefix=" = ";
    }else{
      addins="";
    }
    try{
      String catalogName=conn.getCatalog();
      stmt=conn.createStatement();
      ResultSet relations=null;
      ResultSet check=null;
      try{
        relations=conn.getMetaData().getColumns(catalogName,null,tableName,"%");

        if(!checkIsDone){
          check = conn.getMetaData().getTables(catalogName, null, getLogTableName(), types);
          this.checkIsDone = true;
          this.logTableExists = check.next();
          check.close();
        }
      }catch(UnsupportedOperationException ex){
        String message="Error while trying to get meta data from target table.";
        throw new SQLException(message);
      }

      while(relations.next()){
        if(columnName.equalsIgnoreCase(relations.getString(4))){
//ZK change from CheckType to targetConfigReader 7.5.2004

           try {
            if(configReader.isNumber(dataType)){
                  relValue=prefix+replacement+addins;
               }else{
                 relValue=prefix+"'"+replacement+"'"+addins;
               }
        	}catch (LoaderException e) {
           
            LoaderException le = new LoaderException("Exception:",e);
            throw le;
        	}
          if(this.logTableExists)
          stmt.executeUpdate("INSERT INTO "+getLogTableName()+" ("+
                                             getLogImportDefinitionName()+", "+getLogOperationName()+", "+getLogTypeName()+", "+
                                             getLogTable()+", "+getLogColumnName()+", "+getLogRowNumber()+", "+
                                             getLogOriginalValue()+", "+getLogNewValue()+
                                             ")"+
                                             " VALUES ("+
                                             "'"+importDefinitionName+"','"+typeOfInsert+"','CLEANING RELATIONS','"+
                                             tableName+"','"+columnName+"','"+(rowNumber+1)+"',"+
                                             "null,'"+replacement+"'"+
                                             ")");

          else{
            this.logger.write("full","\t "+getLogTableName()+" does not exists, or is invalid!");
          }
          this.logger.write("full","\tBecause DataCleaning features is turn on, some replacement were made:");
          this.logger.write("full","\t "+typeOfInsert+":RELATIONS: In table "+tableName+", column "+columnName+", value null were replaced" );
          this.logger.write("full","\t with value "+replacement+" (row number "+(rowNumber+1)+")" );

        }
      }
      relations.close();
      stmt.close();
    }catch (SQLException ex){
      this.logger.write("full","\t Error:"+ ex.getMessage());
    }
    return relValue;
  }

  /**
   * This method write message in to log table if some data must be truncated
   * @param tableName is name of the table which is in the process
   * @param conn is connection to target database (table)
   * @param strQuery is string representing sql statement
   * @param rowNumber is current row number
   * @param update is the parameter which may be 'insert' or 'update'
   * @param onErrorContinue is value of onErrorContinue attribute
   * @param importDefinitionName is name of the import definition job
   * @throws SQLException
   */
  public void cutingDataLenghtUpdate(String tableName, Connection conn, String strQuery, int rowNumber,
                                     String update, String onErrorContinue,
                                     String importDefinitionName) throws SQLException, LoaderException{

    boolean end=false;
    boolean endTemp=false;
    SqlParser sqlParser=new SqlParser();

    try{
      try{
        sqlParser.parse(update+strQuery);
      }catch(Exception e){
        throw new SQLException(e.getMessage());
      }
      String[] columnNames=sqlParser.getColumnNames();
      String[] columnValues=sqlParser.getColumnValues();

      String[] types = {"TABLE"};
      String catalogName=conn.getCatalog();
      stmt=conn.createStatement();

      if (!this.currentTableName.equalsIgnoreCase(tableName)) {
        try {
          rsUpdate = conn.getMetaData().getColumns(catalogName, null, tableName,
              "%");
        }
        catch (UnsupportedOperationException ex) {
          String message ="Error while trying to get meta data from target table.";
          throw new SQLException(message);
        }
        colNamesDataTypesUpdate.clear();
        colNamesDataLenghtUpdate.clear();
        this.currentTableName = tableName;
        while (rsUpdate.next()) {
          String columnName = rsUpdate.getString(4).toUpperCase();
          colNamesDataTypesUpdate.put(columnName, rsUpdate.getString(6));
          colNamesDataLenghtUpdate.put(columnName, rsUpdate.getString(7));
        }
        rsUpdate.close();
      }

      for (int i = 0; i < columnNames.length; i++) {
        String dataNameInQyery = columnNames[i];
        String dataValueInQyery = columnValues[i];
        String dataTypeName = (String) colNamesDataTypesUpdate.get(dataNameInQyery.toUpperCase());
        String dataTypeLenght = (String) colNamesDataLenghtUpdate.get(dataNameInQyery.toUpperCase());
        if (!dataValueInQyery.equalsIgnoreCase("null")) {
//			ZK change from CheckType to targetConfigReader 7.5.2004
         try{
          
          if (!configReader.isNumber(dataTypeName)) {
            if ((dataValueInQyery.length()) > Integer.parseInt(dataTypeLenght)) {
              String original = dataValueInQyery.substring(0,dataValueInQyery.length());
              String replacement = replaceQuote(original,Integer.parseInt(dataTypeLenght));
              update = replaceFirst(update, dataValueInQyery, replacement);
              ResultSet check = null;
              try {
                if(!checkIsDone){
                  check = conn.getMetaData().getTables(catalogName, null, getLogTableName(), types);
                  this.checkIsDone = true;
                  this.logTableExists = check.next();
                  check.close();
                }
              }
              catch (UnsupportedOperationException ex) {
                String message ="Error while trying to get meta data from target table.";
                throw new SQLException(message);
              }
              if (this.logTableExists) {
                stmt.executeUpdate("INSERT INTO " + getLogTableName() + " (" +
                                   getLogImportDefinitionName() + ", " +
                                   getLogOperationName() + ", " +
                                   getLogTypeName() + ", " +
                                   getLogTable() + ", " + getLogColumnName() +
                                   ", " + getLogRowNumber() + ", " +
                                   getLogOriginalValue() + ", " +
                                   getLogNewValue() +
                                   ")" +
                                   " VALUES (" +
                                   "'" + importDefinitionName +
                                   "','UPDATE','CUT OFF DATA','" +
                                   tableName + "','" + dataNameInQyery + "','" +
                                   (rowNumber+1) + "','" +
                                   original + "','" + replacement + "'" +
                                   ")");

              }
              else {
                this.logger.write("full","\t " + getLogTableName() + " does not exists, or is invalid");
              }
              this.logger.write("full","\t Because dataCutOff attribute is true, some replacement were made:");
              this.logger.write("full","\t UPDATE:CutOff: In table "+tableName+", column "+dataNameInQyery+", value "+dataValueInQyery+" were replaced" );
              this.logger.write("full","\t with value " + replacement + " (row number " +(rowNumber+1) + ")");

            }
          }
         }catch (LoaderException e){
         	LoaderException le = new LoaderException("Exception:",(Throwable)e);
         	throw le;
         	
         }
        }
      }
      stmt.executeUpdate(update+strQuery);
      stmt.close();

    }catch(SQLException ex){
      if(onErrorContinue.equalsIgnoreCase("true"))
        cleaningInsert(tableName,conn,rowNumber,"UPDATE","Update fails. Message: "+ex.getMessage(),importDefinitionName);
      else{
        LoaderException le = new LoaderException("SQLException: ",(Throwable)ex);
        this.logger.write("full", "\tError in SQL statement: " +le.getCause());
//        this.logger.write("full", "\tError : Because OnErrorContinue Attribute is false, application is terminated");
        throw  ex;

      }
    }
  }

  /**
   * This method write message in to log table if some data must be truncated
   * @param tableName is name of the table which is in the process
   * @param conn is connection to target database (table)
   * @param strQuery is string represented sql statement
   * @param rowNumber is current row number
   * @param onErrorContinue represents onErrorContinue tag
   * @param msg represents error message
   * @param importDefinitionName is name of the import definition job
   * @throws SQLException
   */
  public void cutingDataLenght(String tableName, Connection conn, String strQuery,
                               int rowNumber, String onErrorContinue, String msg,
                               String importDefinitionName)throws SQLException{
    boolean end=false;
    boolean endTemp=false;
    SqlParser sqlParser=new SqlParser();

    try{
      try{
        sqlParser.parse(strQuery);
      }catch(Exception e){
        throw new SQLException(e.getMessage());
      }
      String[] columnNames=sqlParser.getColumnNames();
      String[] columnValues=sqlParser.getColumnValues();

      String[] types = {"TABLE"};
      String catalogName=conn.getCatalog();
      stmt=conn.createStatement();

      if(!this.currentTableName.equalsIgnoreCase(tableName)){
        try{
          rs=conn.getMetaData().getColumns(catalogName,null,tableName,"%");
        }catch(UnsupportedOperationException ex){
          String message="Error while trying to get meta data from target table.";
          throw new SQLException(message);
        }
        colNamesDataTypes.clear();
        colNamesDataLenght.clear();
        this.currentTableName=tableName;
        while (rs.next()){
          String columnName=rs.getString(4).toUpperCase();
          colNamesDataTypes.put(columnName,rs.getString(6));
          colNamesDataLenght.put(columnName,rs.getString(7));
        }
      }

      for (int i = 0; i < columnNames.length; i++) {
        String dataNameInQyery=columnNames[i];
        String dataValueInQyery=columnValues[i];
        String dataTypeName=(String)colNamesDataTypes.get(dataNameInQyery.toUpperCase());
        String dataTypeLenght=(String)colNamesDataLenght.get(dataNameInQyery.toUpperCase());
        if(!dataValueInQyery.equalsIgnoreCase("null")){
//			ZK change from CheckType to targetConfigReader 7.5.2004

         try {
                     if(!configReader.isNumber(dataTypeName)){
                        if((dataValueInQyery.length())>Integer.parseInt(dataTypeLenght)){
                          String original=dataValueInQyery.substring(0,dataValueInQyery.length());
                          String replacement=replaceQuote(original,Integer.parseInt(dataTypeLenght));
                          strQuery=replaceFirst(strQuery ,dataValueInQyery,replacement);
                          ResultSet check=null;
                          try{
                            if(!checkIsDone){
                              check = conn.getMetaData().getTables(catalogName, null, getLogTableName(), types);
                              this.checkIsDone=true;
                              this.logTableExists=check.next();
                              check.close();
                            }
                          }catch(UnsupportedOperationException ex){
                            String message="Error while trying to get meta data from target table.";
                            throw new SQLException(message);
                          }
                          if(this.logTableExists){
                          stmt.executeUpdate("INSERT INTO "+getLogTableName()+" ("+
                                                       getLogImportDefinitionName()+", "+getLogOperationName()+", "+getLogTypeName()+", "+
                                                       getLogTable()+", "+getLogColumnName()+", "+getLogRowNumber()+", "+
                                                       getLogOriginalValue()+", "+getLogNewValue()+
                                                       ")"+
                                                       " VALUES ("+
                                                       "'"+importDefinitionName+"','INSERT','CUT OFF DATA','"+
                                                       tableName+"','"+dataNameInQyery+"','"+(rowNumber+1)+"','"+
                                                       original+"','"+replacement+"'"+
                                                       ")");
            
                          }else{
                            this.logger.write("full","\t "+getLogTableName()+" does not exists, or is invalid");
            
                          }
            //              check.close();
                          this.logger.write("full","\t Because dataCutOff attribute is true, some replacement were made:");
                          this.logger.write("full","\t INSERT:CutOff: In table "+tableName+", column "+dataNameInQyery+", value "+dataValueInQyery+" were replaced" );
                          this.logger.write("full","\t with value "+replacement+" (row number "+(rowNumber+1)+")" );
                        }
                      }
        
        } catch (LoaderException e) {
        	LoaderException le = new LoaderException("Exception:",(Throwable)e);
           
        }
        }
      }

      stmt.executeUpdate(strQuery);
      stmt.close();
    }catch(SQLException ex){
      if(onErrorContinue.equalsIgnoreCase("true"))
        cleaningInsert(tableName,conn,rowNumber,"INSERT",msg, importDefinitionName);
      else{
        LoaderException le = new LoaderException("SQLException: ",(Throwable)ex);
        this.logger.write("full", "\tError in SQL statement: " +le.getCause());
//        this.logger.write("full", "\tError : Because OnErrorContinue Attribute is false, application is terminated");
        throw  ex;

      }
    }
  }

  private String replaceQuote(String replacement, int length){
    replacement=Utils.replaceAll(replacement,"''","'");
    replacement=replacement.substring(0,length);
    int index=replacement.indexOf("'");
    if(index!=-1){
      replacement=Utils.replaceAll(replacement,"'","''");
    }
    return replacement;
  }

  private String replaceFirst(String input, String forReplace, String replaceWith) {
    String retVal = input;
    int start = input.indexOf( forReplace );
    int end = start + forReplace.length();
    if( start != -1 ) {
      retVal = input.substring( 0,start ) + replaceWith + input.substring( end );
    }
    return retVal;
  }

  /**
   * Set Logger object
   * @param logger Logger object which is used for log file
   */
  public void setLogger(Logger logger) {
    this.logger = logger;
  }
}