loader.java

数据仓库工具

	            pstmtKeys.clear();
	            pstmtForInsert.clear();
	            //release transformations
	            List transformations = this.transformations.getTransformations();
	            for(int iCount = 0; iCount < transformations.size(); iCount++ ) {
	            	((Transformation)transformations.get(iCount)).release();
	            }
            }
          } else
            this.logger.write("normal",
                "\tImportDefinition " + importName + " is excluded from process");
          //this.logger.setEnabledLogLevels("none");
          iImport++;
        }
        //rale 24.10.2003
        else if (vecXmlElements.get(l).toString().equalsIgnoreCase("copyTable")) {
          if (importDefinitionElement.bRestartAutoCreate == true) {
            dataTransmition.createRestartCounterTable(connToPools);
          }
          this.currentJobName = "copyTable";
          String importName = importDefinitionElement.strCopyTableName;
          boolean doJob = true;
          for (int i = 0; i < this.includeTables.size(); i++) {
            if (importName.equalsIgnoreCase(this.includeTables.get(i).toString())) {
              doJob = true;
              break;
            } else {
              doJob = false;
            }
          }
          if (doJob) {
            try {
              this.loadSourceCT(connToPools);
            }
            catch (Exception e) {
              //this.logger.write("normal", "\t" + e.getMessage());
              LoaderException le = new LoaderException("SQLException: ", (Throwable)e);
							this.logger.write("full", le.getStackTraceAsString());
              if (importDefinitionElement.strCopyTableOnErrorContinue.equalsIgnoreCase("true")) {
                this.logger.write("normal",
                    "\tError : Application is not terminated because 'OnErrorContinue' attribute has value 'true'");
              } else {
                this.logger.write("normal",
                    "\tError : Application is terminated because 'OnErrorContinue' attribute has value 'false'");
                connManager.release("true");
                throw le;
              }
            }
          } else {
            this.logger.write("normal", "\tCopyTable " + importName + " is excluded from process");
          }
//          this.logger.setEnabledLogLevels(loaderJobReader.getDefaultLogMode());
          iCopy++;
        } else if (vecXmlElements.get(l).toString().equalsIgnoreCase("echo")) {
          this.currentJobName = "echo";
          logger.writeEcho(this.logger.getMessage(new Integer(iEcho).toString()));
          iEcho++;
        } else if (vecXmlElements.get(l).toString().equalsIgnoreCase("sql")) {
          this.currentJobName = "sql";
          String sqlName = sqlElement.getSqlName();
          boolean doJob = true;
          for (int i = 0; i < this.includeTables.size(); i++) {
            if (sqlName.equalsIgnoreCase(this.includeTables.get(i).toString())) {
              doJob = true;
              break;
            } else {
              doJob = false;
            }
          }
          if (doJob) {
            try {
              if (sqlElement.getSqlCommit().equalsIgnoreCase("true")) {
                connToPools.setAutoCommit(true);
//ZK change this because of problems with onErrorContinue="true" and commit="true" in <sql ...>  tag
                try{
                    this.executeSQLStatement(connToPools);
                }catch(Exception e){
                	throw e;
                }
                finally{
					connToPools.setAutoCommit(false);
                }
              } else
                this.executeSQLStatement(connToPools);
            }
            catch (Exception e) {
              LoaderException le = new LoaderException("SQLException: ",
                                   (Throwable)e);
              //this.logger.write("normal", "\t" + e.getMessage());
							this.logger.write("full", le.getStackTraceAsString());
              if (sqlElement.getSqlOnErrorContinue().equalsIgnoreCase("true")) {
                this.logger.write("normal",
                    "\tError : Because OnErrorContinue Attribute is true application is not terminated");
              }
              else if (sqlElement.getSqlOnErrorContinue().equalsIgnoreCase("")&&
                  loaderJobReader.getDefaultOnErrorContinue()==true) {

                this.logger.write("normal",
                    "\tError : Because OnErrorContinue Attribute is true application is not terminated");
              }
              else {
                this.logger.write("normal",
                    "\tError : Because OnErrorContinue Attribute is false application is terminated");
                connManager.release("true");
                throw le;
              }
            }
          } else {
            this.logger.write("normal", "\tSql statement " + sqlName + " is excluded from process");
          }
          iSql++;
//          this.logger.setEnabledLogLevels(loaderJobReader.getDefaultLogMode());
        } else {
          //this.logger.write("normal",
          //    "Error : Loader supports loading data up to maximum 3 SQL database");
          LoaderException le = new LoaderException("SQLException: ",
                               (Throwable) (new SQLException(
                               "Error : Loader supports loading data up to maximum 3 SQL database")));
		      this.logger.write("full", le.getStackTraceAsString());
          if (connToPools != null)
            connManager.release("true");
          throw le;
        }
        
        if (!this.hasRestartCounter && iLastImportDefinition == l) {
          if (connToPools!=null && !connToPools.isClosed())
		  		 
            connToPools.commit();
        }
       
      }
	   this.logger.setEnabledLogLevels(this.defaultLogMode);
	   connManager.release("false");
		this.logger.write("normal", "All rows are commited.");
		this.logger.write("normal", "Application is finished.");
		this.logger.write("normal", "All jobs duration: " + this.timeCounter.getTotalTime());
    }
    catch (IOException e) {
      LoaderException le = new LoaderException("IOException: ", (Throwable)e);
      //this.logger.write("normal", e.getMessage());
	  	this.logger.write("full", le.getStackTraceAsString());
      throw le;
    }
    catch (ClassNotFoundException e) {
      LoaderException le = new LoaderException("ClassNotFoundException: ",
                           (Throwable)e);
      //this.logger.write("normal", le.getCause().toString());
	  	this.logger.write("full", le.getStackTraceAsString());
      
      throw le;
    }
    catch (SQLException e) {
      LoaderException le = new LoaderException("SQLException: ", (Throwable)e);
      //this.logger.write("normal", le.getCause().toString());
	  	this.logger.write("full", le.getStackTraceAsString());
      throw le;
    }
    catch (NullPointerException e) {
      LoaderException le = new LoaderException("NullPointerException: " +
                           e.getMessage(),
                           (Throwable)e);
      //this.logger.write("normal", le.getCause().toString());
			this.logger.write("full", le.getStackTraceAsString());
      throw le;
    }
    catch (SAXException e) {
      LoaderException le = new LoaderException("SAXException: ", (Throwable)e);
      //this.logger.write("normal", le.getCause().toString());
			this.logger.write("full", le.getStackTraceAsString());
      throw le;
    }
    catch (Exception e) {
      LoaderException le = new LoaderException("Exception: ", (Throwable)e);
      //this.logger.write("normal", le.getCause().toString());
			this.logger.write("full", le.getStackTraceAsString());
      throw le;
    }
    finally {
      try {
        if (connToPools != null && !connToPools.isClosed())
          connToPools.close();
        this.logger.close();
      }
      catch (SQLException e) {
        LoaderException le = new LoaderException("SQLException: ",
                             (Throwable)e);
        //this.logger.write("normal", le.getCause().toString());
				this.logger.write("full", le.getStackTraceAsString());
        throw le;
      }
    }
  }

  /**
   * Method loadSourceCT is used to load data from the source table (Csv, MySql, MsSql, Exel...)
   * into target tables.
   * @param connTarget is connection object to target database
   * @throws java.lang.Exception
   */
  private void loadSourceCT(Connection connTarget) throws Exception {

    this.timeCounter.setStartJobTime();
    DateFormat dfNow = DateFormat.getDateTimeInstance();
    Date currentDate = new Date();

    this.logger.write("full", "\tloadSource method is started.");
    this.logger.write("normal", "Copy Table " +
        importDefinitionElement.strCopyTableName
        + " is started at " + dfNow.format(currentDate) + ".");

    Connection connSource = null;
    ResultSet rsetSource = null;
    ResultSet rsetTarget = null;
    Statement stmtSource = null;
    Statement stmtTarget = null;

    String strQuery = "";
    Vector vecNewColumnNames = new Vector();
    Vector vecNewColumnTypes = new Vector();
    Hashtable tmpColumnReader = new Hashtable();

    boolean bOidLogicCurrentTable = new Boolean(importDefinitionElement.strCopyTableOidLogic).
                                    booleanValue();
    boolean isTOS = ! (importDefinitionElement.strObjectIDNameColumnName.equals(
                    "") || importDefinitionElement.strObjectIDNameColumnValue.equals(""));
    try {
      this.checkOidLogic(connTarget);
      if (configReaderSource.getRequiredUser()) {
        if (jdbcParametersElement.getJDBCSourceParameterUser().equals(""))
          this.inputUser(true);
        Class.forName(jdbcParametersElement.getJDBCSourceParameterDriver());
        connSource = DriverManager.getConnection(jdbcParametersElement.
                     getJDBCSourceParameterConnection(),
                     jdbcParametersElement.
                     getJDBCSourceParameterUser(),
                     jdbcParametersElement.
                     getJDBCSourceParameterPassword());
      } else {
        Class.forName(jdbcParametersElement.getJDBCSourceParameterDriver());
        connSource = DriverManager.getConnection(jdbcParametersElement.
                     getJDBCSourceParameterConnection());
      }

      int iRowNumber = 0;
      boolean isEndFlag = false;
      stmtSource = connSource.createStatement();
      stmtTarget = connTarget.createStatement();

      //setCursorName
      boolean setCursorName = false;
      if (!importDefinitionElement.setCursorNameCT.equalsIgnoreCase(""))
        setCursorName = new Boolean(importDefinitionElement.setCursorNameCT).booleanValue();
      else
        setCursorName = configReaderSource.getSetCursorNameEnabled();

      if (setCursorName) {
        long cursorName = System.currentTimeMillis();
        stmtSource.setCursorName(String.valueOf(cursorName));
      }
      //setFetchSize
      boolean setFetchSize = false;
      if (!importDefinitionElement.setFetchSizeCT.equalsIgnoreCase(""))
        setFetchSize = new Boolean(importDefinitionElement.setFetchSizeCT).booleanValue();
      else
        setFetchSize = configReaderSource.getSetFetchSizeEnabled();

      if (setFetchSize)
        stmtSource.setFetchSize(importDefinitionElement.iCopyTableCommitCount);
//ZK added this	5.4 2004
//begin
//next line will give me parameter getColumnsSupported from *conf.xml. It describe is getColumns() method is suported
//by used driver

	   boolean columnsSuportedTarget = configReaderTarget.getColumnsSupported();
	   boolean columnsSuportedSource = configReaderSource.getColumnsSupported();

	   strQuery = "select * from ";

//	   sourceColumnCount is used out of try block also
		rsetSource = stmtSource.executeQuery(strQuery + importDefinitionElement.strCopyTableSourceTableName);
	   int sourceColumnCount = rsetSource.getMetaData().getColumnCount();
	   try {
		  if (columnsSuportedSource){
	        ResultSet rsetSourceTemp = connSource.getMetaData().
	      	getColumns( connSource.getCatalog(), null, importDefinitionElement.strCopyTableSourceTableName, "%" );
			  while(rsetSourceTemp.next()){
					vecNewColumnNames.add(rsetSourceTemp.getString(4));
			  }
			rsetSourceTemp.close();

	     }else{
				//rsetSource and source column count are already maked above try block
			      for (int i = 1; i < sourceColumnCount + 1; i++) {
								vecNewColumnNames.add(rsetSource.getMetaData().getColumnName(i));
		    		}
		 }

		 if (columnsSuportedTarget){		 	
	  		ResultSet rsetTargetTemp = connTarget.getMetaData().
				getColumns( connTarget.getCatalog(), null, importDefinitionElement.strCopyTableTargetTableName, "%" );

			while (rsetTargetTemp.next()) {
								String targetColumnName = rsetTargetTemp.getString(4);
								String targetColumnType = rsetTargetTemp.getString(6);
								tmpColumnReader.put(targetColumnName.toUpperCa