databasemigration.java

一个java的ETL工具

		Element connectionSource_EL = (Element)xmlRoot.getElementsByTagName("CONNECTION_SOURCE").item(0);  // there is one and only one named CONNECTION_SOURCE Element
		sourceDB_Driver = connectionSource_EL.getAttribute("DRIVER");
		sourceDB_URL = connectionSource_EL.getAttribute("URL");
		sourceDB_Username = connectionSource_EL.getAttribute("USERNAME");
		sourceDB_Password = connectionSource_EL.getAttribute("PASSWORD");

		Element connectionTarget_EL = (Element)xmlRoot.getElementsByTagName("CONNECTION_TARGET").item(0);  // there is one and only one named CONNECTION_TARGET Element
		targetDB_Driver = connectionTarget_EL.getAttribute("DRIVER");
		targetDB_URL = connectionTarget_EL.getAttribute("URL");
		targetDB_Username = connectionTarget_EL.getAttribute("USERNAME");
		targetDB_Password = connectionTarget_EL.getAttribute("PASSWORD");

		if (debug || operation.equals("TEST"))
		{
			System.out.println("\n");

			System.out.println("SOURCE CONNECTION:");			
			System.out.println("Driver = " + sourceDB_Driver);			
			System.out.println("URL = " + sourceDB_URL);			
			System.out.println("Username = " + sourceDB_Username);			
			System.out.println("Password = " + sourceDB_Password);			

			System.out.println("\n");
						
			System.out.println("TARGET CONNECTION:");			
			System.out.println("Driver = " + targetDB_Driver);			
			System.out.println("URL = " + targetDB_URL);			
			System.out.println("Username = " + targetDB_Username);			
			System.out.println("Password = " + targetDB_Password);			
	
			System.out.println("\n");
		}

	
		if (sourceDB_Driver.equals("") || sourceDB_URL.equals("") || targetDB_Driver.equals("") || targetDB_URL.equals(""))
		{
			System.out.println("Error in XML file:  Not enough DB Connection Info");
			System.exit(0);
		}
		else
		{
			DatabaseMetaData dbm;
			
			try
			{
				//  establish source connection
				Class sourceClass = Class.forName(sourceDB_Driver);
				System.out.println ("Loaded " + sourceClass);

				sourceConnection = DriverManager.getConnection(sourceDB_URL, sourceDB_Username, sourceDB_Password);
				sourceConnection.setReadOnly(true);

				dbm = sourceConnection.getMetaData();
				System.out.println(dbm.getDatabaseProductName());
				System.out.println("Source Connection OK");

				//  establish target connection
				Class targetClass = Class.forName(targetDB_Driver);
				System.out.println ("Loaded " + targetClass);

				targetConnection = DriverManager.getConnection(targetDB_URL, targetDB_Username, targetDB_Password);

				dbm = targetConnection.getMetaData();
				System.out.println(dbm.getDatabaseProductName());
				System.out.println("Target Connection OK");
				System.out.println();

				connected = true;
			}
			catch (Exception e)
			{
				System.out.println("Database Connection Failed");
				System.out.println("\n");
				System.out.println(e);
				System.exit(0);
			}
		}
	}

	//  Migration Methods

	private void migrate() throws Exception
	{
System.out.println("migrate:  start");

		for (int i=0; i<migration_AL.size(); i++)
		{
System.out.println("migration # " + i);
			Migration migration = (Migration)migration_AL.get(i);
System.out.println(migration);
			
			migration.sortTransfers();

			// build Source Stuff
			
System.out.println("build Source Stuff");
			Statement sourceStatement = sourceConnection.createStatement(ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_READ_ONLY);
			String sourceSELECT = migration.createSourceSQL();
System.out.println("Source SQL:  " + sourceSELECT);
			
			ArrayList sourceFields_AL = migration.getSourceFields();
			
			ArrayList sourceFieldIds = new ArrayList();
			ArrayList sourceFieldNames = new ArrayList();
			ArrayList sourceFieldTypes = new ArrayList();
			
			Hashtable sourceValues_HT;
			for (int j=0; j<sourceFields_AL.size(); j++)
			{
				Field tempField = (Field)sourceFields_AL.get(j);
				
				sourceFieldIds.add(tempField.getId());
				sourceFieldNames.add(tempField.getNameSQL());
				sourceFieldTypes.add(tempField.getTypeSQL());
			}

			// build Target Stuff
						
System.out.println("build Target Stuff");
			String targetINSERT = migration.createTargetSQL_INSERT();
System.out.println("Target INSERT SQL:  " + targetINSERT);
			PreparedStatement targetStatement_INSERT = targetConnection.prepareStatement(targetINSERT);

			ArrayList targetFields_AL = migration.getTargetFields();
			
			ArrayList targetFieldIds = new ArrayList();
			ArrayList targetFieldNames = new ArrayList();
			ArrayList targetFieldTypes = new ArrayList();
			
			Hashtable targetValues_HT;
			for (int j=0; j<targetFields_AL.size(); j++)
			{
				Field tempField = (Field)targetFields_AL.get(j);
				
				targetFieldIds.add(tempField.getId());
				targetFieldNames.add(tempField.getNameSQL());
				targetFieldTypes.add(tempField.getTypeSQL());
			}
			
			// build Temporary Variables
						
System.out.println("build Temporary Variables");
			ArrayList temporaryVariables_AL = migration.getTemporaryVariables();
System.out.println("Temporary Variables Found = " + temporaryVariables_AL.size());			
			ArrayList temporaryVariableIds = new ArrayList();
			ArrayList temporaryVariableTypes = new ArrayList();
			
			Hashtable temporaryValues_HT;
			for (int j=0; j<temporaryVariables_AL.size(); j++)
			{
System.out.println("tv: j = " + j);
				Variable tempVariable = (Variable)temporaryVariables_AL.get(j);
				
				temporaryVariableIds.add(tempVariable.getId());
				temporaryVariableTypes.add(tempVariable.getTypeSQL());
			}
			
			//  retrieve UpdateFallbackRule stuff
			
			UpdateFallbackRule updateFallbackRule = migration.getUpdateFallbackRule();
			ArrayList targetUpdateFields = updateFallbackRule.getTargetUpdateFields();
			
			
			ArrayList transfer_AL = migration.getTransfers();

			//  Read in Each Row of Source Table

System.out.println("Selecting RecordSet from Source");
			ResultSet sourceRS = sourceStatement.executeQuery(sourceSELECT);
			sourceRS.beforeFirst();
						
			while (sourceRS.next())
			{
				System.out.println("Record: " + sourceRS.getRow());

				sourceValues_HT = new Hashtable();				
				targetValues_HT = new Hashtable();				
				temporaryValues_HT = new Hashtable();				

				for (int j=0; j<sourceFields_AL.size(); j++)
				{
					Field field = (Field)sourceFields_AL.get(j);
	
					String id = field.getId();
					String name = field.getNameSQL();
					String type = field.getTypeSQL();

					System.out.print("Field " + j + "(" + id + "): " + name + "(" + type + ") = ");

					try
					{
						if (type.equals("BIT"))
							sourceValues_HT.put(id, new Boolean(sourceRS.getBoolean(name)));
						else if (type.equals("TINYINT"))
							sourceValues_HT.put(id, new Byte(sourceRS.getByte(name)));
						else if (type.equals("SMALLINT"))
							sourceValues_HT.put(id, new Short(sourceRS.getShort(name)));
						else if (type.equals("INTEGER"))
							sourceValues_HT.put(id, new Integer(sourceRS.getInt(name)));
						else if (type.equals("BIGINT"))
							sourceValues_HT.put(id, new Long(sourceRS.getLong(name)));
						else if (type.equals("REAL"))
							sourceValues_HT.put(id, new Float(sourceRS.getFloat(name)));
						else if (type.equals("DOUBLE"))
							sourceValues_HT.put(id, new Double(sourceRS.getDouble(name)));
						else if (type.equals("CHAR"))
							sourceValues_HT.put(id, sourceRS.getString(name));
						else if (type.equals("VARCHAR"))
							sourceValues_HT.put(id, sourceRS.getString(name));
						else if (type.equals("LONGVARCHAR"))
							sourceValues_HT.put(id, sourceRS.getString(name));
						else if (type.equals("DATE"))
							sourceValues_HT.put(id, sourceRS.getDate(name));
						else if (type.equals("TIME"))
							sourceValues_HT.put(id, sourceRS.getTime(name));
						else if (type.equals("TIMESTAMP") || type.equals("DATETIME"))
							sourceValues_HT.put(id, sourceRS.getTimestamp(name));
						else if (type.equals("BLOB"))
							sourceValues_HT.put(id, sourceRS.getObject(name));
	
						System.out.println(sourceValues_HT.get(id));
					}
					catch (Exception e)
					{
						System.out.println("null (DB has null value)");
					}	
				}
						
				// review each transfer for modifications to result set & place into target hash table

				boolean doTransfer = false;
				System.out.println("Checking Go/NoGo Transfer Conditions");
				for (int j=0; j<transfer_AL.size(); j++)   
				{
					Transfer transfer = (Transfer)transfer_AL.get(j);

					boolean conditionResult = true;
					Condition condition = transfer.getCondition();
					if (condition != null)
					{
						ArrayList comparisons = condition.getComparisons();
						for (int k=0; k<comparisons.size(); k++)
						{
							Comparison comparison = (Comparison)comparisons.get(k);
							
							String relationToPrevious = comparison.getRelationToPrevious();
							String firstOperand = comparison.getFirstOperand();
							String operator = comparison.getOperator();
							String secondOperand = comparison.getSecondOperand();
							
							boolean comparisonResult = false;
							
							//  XML Logical Validation  -- should be done in buildMigration() instead --  on read of XML
							//  assume legal values for relationToPrecious and operator, since ATTRIBUTES are validated
							
							boolean comparisonLogicallyValid = true;
							
							int firstField_i = sourceFieldIds.indexOf(firstOperand);
							if (firstField_i == -1)
							{
								System.out.println("Source Id NOT FOUND for Comparison #" + k);
								comparisonLogicallyValid = false;
							}
							  
							int secondField_i = sourceFieldIds.indexOf(secondOperand);
							if ( (secondField_i == -1) && (!secondOperand.startsWith("\"") && !secondOperand.endsWith("\"")) )
							{
								System.out.println("Target Id OR String Literal NOT FOUND for Comparison #" + k);
								comparisonLogicallyValid = false;
							}	
							
							if ( (k>0) && relationToPrevious.equals("") ) 
							{
								System.out.println("Relation to Previous NOT FOUND for Comparison #" + k);
								comparisonLogicallyValid = false;
							}	

							if (!comparisonLogicallyValid)
							{
								System.out.println("Illegal Comparison #" + k);
							}
							else
							{
								if ( ((String)sourceFieldTypes.get(firstField_i)).endsWith("CHAR") )  // String Comparisons
								{
									String firstOperandValue = (String)sourceValues_HT.get(firstOperand);
									String secondOperandValue;
									if (secondField_i == -1)  // must be a literal
										secondOperandValue = secondOperand.substring(1,secondOperand.length()-1);
									else
										secondOperandValue = (String)sourceValues_HT.get(secondOperand);
									
									if ((firstOperandValue != null) && (secondOperandValue != null))		
									{
										if (operator.equals("EQ"))	if (operator.equals("EQ"))
											comparisonResult = firstOperandValue.equals(secondOperandValue);
										else if (operator.equals("NE"))	
											comparisonResult = !firstOperandValue.equals(secondOperandValue);
										else if (operator.equals("IN"))
											comparisonResult = (firstOperand.indexOf(secondOperand) >= 0); 	
									}
									else
									{
										comparisonResult = false;
									}				
								}
								else if ( ((String)sourceFieldTypes.get(firstField_i)).equals("DATE") )  // Date Comparisons
								{
									comparisonResult = true;
								}
								else
								{
									comparisonResult = false;
								}	
							}		
						
							if (k==0)  //  First Comparison
								conditionResult = comparisonResult;
							else
							{
								if (relationToPrevious.equals("AND"))
									conditionResult = (conditionResult && comparisonResult);
								else if (relationToPrevious.equals("INOR"))
									conditionResult = (conditionResult || comparisonResult);
								else if (relationToPrevious.equals("AND"))
									if (conditionResult != comparisonResult)
										conditionResult = false;
									else
										conditionResult = false; 	
								else
									conditionResult = false;
							}		
						}
					}	
						
					if (conditionResult == false) 
					{
						System.out.println("Source Row FAILED Condition - not transfered.");
						break;  // to next row source
					}

					// perform Actions
					
					// NOTE: no type checking done
					
					doTransfer = true;  //  at least one Transfer has met its Condition(s)
					
					System.out.println("Processing Actions");
					
					ArrayList actions = transfer.getActions();
					for (int k=0; k<actions.size(); k++)
					{
						Action action = (Action)actions.get(k);
						
						System.out.println(action.getClass());
						String type = action.getType();
						
						if (type.equals("PUSH"))
						{
							String fromId = ((Push)action).getFromId();
							String toId = ((Push)action).getToId();
							
							//  find Ids
							
							boolean validIds = true;
							
							int fromId_i = -1;
							int toId_i = -1;
							String fromId_ht = "";
							String toId_ht = "";
							
							fromId_i = sourceFieldIds.indexOf(fromId);
							if (fromId_i != -1)
								fromId_ht = "source";
							else
							{
								fromId_i = targetFieldIds.indexOf(fromId);
								if (fromId_i != -1)
									fromId_ht ="target";
								else
								{						
									fromId_i = temporaryVariableIds.indexOf(fromId);
									if (fromId_i != -1)
										fromId_ht ="temporary";
									else
										validIds = false;
								}
							}
							
							toId_i = sourceFieldIds.indexOf(toId);
							if (toId_i != -1)
								toId_ht = "source";
							else
							{
								toId_i = targetFieldIds.indexOf(toId);
								if (toId_i != -1)
									toId_ht ="target";
								else
								{						
									toId_i = temporaryVariableIds.indexOf(toId);
									if (toId_i != -1)
										toId_ht ="temporary";
									else
										validIds = false;
								}
							}
							
//System.out.println("PUSH: "+fromId+"["+fromId_ht+"("+fromId_i+")]-->"+toId+"["+toId_ht+"("+toId_i+")]");
							
							if (validIds)
							{
								Object pushObject = null;
								
								// pull data
								
								if (fromId_ht.equals("source"))
									pushObject = sourceValues_HT.get(fromId);
								else if (fromId_ht.equals("target"))