	public static List<CleanPlan> createCleanPlanFromJSON(Reader reader)
			throws InvalidRuleException, InvalidCleanPlanException {
		Preconditions.checkNotNull(reader);
		JSONObject jsonObject = (JSONObject) JSONValue.parse(reader);
		List<CleanPlan> result = Lists.newArrayList();
		boolean isCSV = false;
		List<Schema> schemas = Lists.newArrayList();
		Connection conn = null;
		try {
			JSONObject src = (JSONObject) jsonObject.get("source");
			String type = (String) src.get("type");
			DBConfig dbConfig;
			switch (type) {
			case "csv":
				isCSV = true;
				dbConfig = NadeefConfiguration.getDbConfig();
				break;
			default:
				SQLDialect sqlDialect = SQLDialect.POSTGRES;
				DBConfig.Builder builder = new DBConfig.Builder();
				dbConfig =
                    builder.username((String) src.get("username"))
						.password((String) src.get("password"))
						.url((String) src.get("url")).dialect(sqlDialect)
						.build();
			}
			DBConnectionFactory.initializeSource(dbConfig);
			JSONArray ruleArray = (JSONArray) jsonObject.get("rule");
			ArrayList<Rule> rules = Lists.newArrayList();
			List<String> targetTableNames;
            List<String> fileNames = Lists.newArrayList();
			for (int i = 0; i < ruleArray.size(); i++) {
				schemas.clear();
                fileNames.clear();
				JSONObject ruleObj = (JSONObject) ruleArray.get(i);
				if (isCSV) {
					List<String> fullFileNames = (List<String>) src.get("file");
                    for (String fullFileName : fullFileNames) {
                        fileNames.add(Files.getNameWithoutExtension(fullFileName));
                    }
					if (ruleObj.containsKey("table")) {
						targetTableNames = (List<String>) ruleObj.get("table");
						Preconditions.checkArgument(
					        targetTableNames.size() <= 2,
							"NADEEF only supports MAX 2 tables per rule."
                        );
                        for (String targetTableName : targetTableNames) {
                            if (!fileNames.contains(targetTableName)) {
                                throw new InvalidCleanPlanException("Unknown table name.");
                            }
                        }
					} else {
						targetTableNames = Lists.newArrayList();
						for (String fileName : fileNames) {
							targetTableNames.add(fileName);
                            if (targetTableNames.size() == 2) {
                                break;
                            }
						}
					}
					conn = DBConnectionFactory.getNadeefConnection();
                    HashSet<String> copiedTables = Sets.newHashSet();
                    for (int j = 0; j < targetTableNames.size(); j++) {
						File file = CommonTools.getFile(fullFileNames.get(j));
                        if (!copiedTables.contains(targetTableNames.get(j))) {
                            String tableName = CSVDumper.dump(conn, file, targetTableNames.get(j));
                            targetTableNames.set(j, tableName);
                            schemas.add(DBMetaDataTool.getSchema(tableName));
                        }
					}
				} else {
					List<String> sourceTableNames = (List<String>) ruleObj.get("table");
					for (String tableName : sourceTableNames) {
						if (!DBMetaDataTool.isTableExist(tableName)) {
                            throw new InvalidCleanPlanException(
                                "The specified table " +
                                tableName +
                                " cannot be found in the source database.");
						}
					}
					if (ruleObj.containsKey("target")) {
						targetTableNames = (List<String>) ruleObj.get("target");
					} else {
						targetTableNames = Lists.newArrayList();
						for (String sourceTableName : sourceTableNames) {
							targetTableNames.add(sourceTableName + "_copy");
						}
					}
					Preconditions.checkArgument(
                        sourceTableNames.size() == targetTableNames.size() &&
                        sourceTableNames.size() <= 2 &&
                        sourceTableNames.size() >= 1,
                        "Invalid Rule property, rule needs to have one or two tables.");
					for (int j = 0; j < sourceTableNames.size(); j++) {
						DBMetaDataTool.copy(
                            sourceTableNames.get(j),
						    targetTableNames.get(j)
                        );
                        schemas.add(
                            DBMetaDataTool.getSchema(
                                targetTableNames.get(j)
                            )
                        );
					}
				}
				type = (String) ruleObj.get("type");
				Rule rule;
				JSONArray value;
				value = (JSONArray) ruleObj.get("value");
				String ruleName = (String) ruleObj.get("name");
                if (Strings.isNullOrEmpty(ruleName)) {
                    ruleName = "Rule" + CommonTools.toHashCode((String)value.get(0));
                }
				switch (type) {
				case "udf":
					value = (JSONArray) ruleObj.get("value");
					Class udfClass =
                        CommonTools.loadClass((String) value.get(0));
					if (!Rule.class.isAssignableFrom(udfClass)) {
						throw new InvalidRuleException(
						    "The specified class is not a Rule class."
                        );
					}
					rule = (Rule) udfClass.newInstance();
					rule.initialize(ruleName, targetTableNames);
					rules.add(rule);
					break;
				default:
					RuleBuilder ruleBuilder = NadeefConfiguration.tryGetRuleBuilder(type);
					if (ruleBuilder != null) {
						rules.addAll(
                            ruleBuilder.name(ruleName)
                                .schema(schemas)
								.table(targetTableNames)
                                .value(value)
                                .build()
                        );
					} else {
						tracer.err("Unknown Rule type: " + type, null);
					}
					break;
				}
			}
			for (int i = 0; i < rules.size(); i++) {
				result.add(new CleanPlan(dbConfig, rules.get(i)));
			}
			return result;
		} catch (Exception ex) {
			ex.printStackTrace();
            if (ex instanceof InvalidRuleException) {
				throw (InvalidRuleException) ex;
			}
			throw new InvalidCleanPlanException(ex);
		} finally {
			if (conn != null) {
				try {
					conn.close();
				} catch (SQLException ex) {
				}
			}
		}
	}

	public static List<CleanPlan> createCleanPlanFromJSON(Reader reader)
			throws InvalidRuleException, InvalidCleanPlanException {
		Preconditions.checkNotNull(reader);
		JSONObject jsonObject = (JSONObject) JSONValue.parse(reader);
		List<CleanPlan> result = Lists.newArrayList();
		boolean isCSV = false;
		List<Schema> schemas = Lists.newArrayList();
		Connection conn = null;
		try {
			JSONObject src = (JSONObject) jsonObject.get("source");
			String type = (String) src.get("type");
			DBConfig dbConfig;
			switch (type) {
			case "csv":
				isCSV = true;
				dbConfig = NadeefConfiguration.getDbConfig();
				break;
			default:
				SQLDialect sqlDialect = SQLDialect.POSTGRES;
				DBConfig.Builder builder = new DBConfig.Builder();
				dbConfig =
                    builder.username((String) src.get("username"))
						.password((String) src.get("password"))
						.url((String) src.get("url")).dialect(sqlDialect)
						.build();
			}
			DBConnectionFactory.initializeSource(dbConfig);
			JSONArray ruleArray = (JSONArray) jsonObject.get("rule");
			ArrayList<Rule> rules = Lists.newArrayList();
			List<String> targetTableNames;
            List<String> fileNames = Lists.newArrayList();
            HashSet<String> copiedTables = Sets.newHashSet();
			for (int i = 0; i < ruleArray.size(); i++) {
				schemas.clear();
                fileNames.clear();
				JSONObject ruleObj = (JSONObject) ruleArray.get(i);
				if (isCSV) {
					List<String> fullFileNames = (List<String>) src.get("file");
                    for (String fullFileName : fullFileNames) {
                        fileNames.add(Files.getNameWithoutExtension(fullFileName));
                    }
					if (ruleObj.containsKey("table")) {
						targetTableNames = (List<String>) ruleObj.get("table");
						Preconditions.checkArgument(
					        targetTableNames.size() <= 2,
							"NADEEF only supports MAX 2 tables per rule."
                        );
                        for (String targetTableName : targetTableNames) {
                            if (!fileNames.contains(targetTableName)) {
                                throw new InvalidCleanPlanException("Unknown table name.");
                            }
                        }
					} else {
						targetTableNames = Lists.newArrayList();
						for (String fileName : fileNames) {
							targetTableNames.add(fileName);
                            if (targetTableNames.size() == 2) {
                                break;
                            }
						}
					}
					conn = DBConnectionFactory.getNadeefConnection();
                    for (int j = 0; j < targetTableNames.size(); j++) {
						File file = CommonTools.getFile(fullFileNames.get(j));
                        if (!copiedTables.contains(targetTableNames.get(j))) {
                            String tableName = CSVDumper.dump(conn, file, targetTableNames.get(j));
                            targetTableNames.set(j, tableName);
                            schemas.add(DBMetaDataTool.getSchema(tableName));
                            copiedTables.add(targetTableNames.get(j));
                        }
					}
				} else {
					List<String> sourceTableNames = (List<String>) ruleObj.get("table");
					for (String tableName : sourceTableNames) {
						if (!DBMetaDataTool.isTableExist(tableName)) {
                            throw new InvalidCleanPlanException(
                                "The specified table " +
                                tableName +
                                " cannot be found in the source database.");
						}
					}
					if (ruleObj.containsKey("target")) {
						targetTableNames = (List<String>) ruleObj.get("target");
					} else {
						targetTableNames = Lists.newArrayList();
						for (String sourceTableName : sourceTableNames) {
							targetTableNames.add(sourceTableName + "_copy");
						}
					}
					Preconditions.checkArgument(
                        sourceTableNames.size() == targetTableNames.size() &&
                        sourceTableNames.size() <= 2 &&
                        sourceTableNames.size() >= 1,
                        "Invalid Rule property, rule needs to have one or two tables.");
					for (int j = 0; j < sourceTableNames.size(); j++) {
						DBMetaDataTool.copy(
                            sourceTableNames.get(j),
						    targetTableNames.get(j)
                        );
                        schemas.add(
                            DBMetaDataTool.getSchema(
                                targetTableNames.get(j)
                            )
                        );
					}
				}
				type = (String) ruleObj.get("type");
				Rule rule;
				JSONArray value;
				value = (JSONArray) ruleObj.get("value");
				String ruleName = (String) ruleObj.get("name");
                if (Strings.isNullOrEmpty(ruleName)) {
                    ruleName = "Rule" + CommonTools.toHashCode((String)value.get(0));
                }
				switch (type) {
				case "udf":
					value = (JSONArray) ruleObj.get("value");
					Class udfClass =
                        CommonTools.loadClass((String) value.get(0));
					if (!Rule.class.isAssignableFrom(udfClass)) {
						throw new InvalidRuleException(
						    "The specified class is not a Rule class."
                        );
					}
					rule = (Rule) udfClass.newInstance();
					rule.initialize(ruleName, targetTableNames);
					rules.add(rule);
					break;
				default:
					RuleBuilder ruleBuilder = NadeefConfiguration.tryGetRuleBuilder(type);
					if (ruleBuilder != null) {
						rules.addAll(
                            ruleBuilder.name(ruleName)
                                .schema(schemas)
								.table(targetTableNames)
                                .value(value)
                                .build()
                        );
					} else {
						tracer.err("Unknown Rule type: " + type, null);
					}
					break;
				}
			}
			for (int i = 0; i < rules.size(); i++) {
				result.add(new CleanPlan(dbConfig, rules.get(i)));
			}
			return result;
		} catch (Exception ex) {
			ex.printStackTrace();
            if (ex instanceof InvalidRuleException) {
				throw (InvalidRuleException) ex;
			}
			throw new InvalidCleanPlanException(ex);
		} finally {
			if (conn != null) {
				try {
					conn.close();
				} catch (SQLException ex) {
				}
			}
		}
	}
