public class FederationSail extends SailBase {
	
	private static final Logger LOGGER = LoggerFactory.getLogger(FederationSail.class);
	
	private List<Repository> members;
	private SourceSelector selector;
	private QueryOptimizer optimizer;
	private EvaluationStrategy evalStrategy;

	private boolean initialized = false;
	
	public FederationSail() {
		this.members = new ArrayList<Repository>();
	}
	
	public void addMember(Repository rep) {
		if (rep == null)
			throw new IllegalArgumentException("federation member must not be NULL");
		this.members.add(rep);
	}
	
	// --- GETTER --------------------------------------------------------------
	
	public EvaluationStrategy getEvalStrategy() {
		return this.evalStrategy;
	}
	
	public QueryOptimizer getFederationOptimizer() {
		return this.optimizer;
	}
	
	public List<Repository> getMembers() {
		return this.members;
	}
	
	public SourceSelector getSourceSelector() {
		return this.selector;
	}
	
	// --- SETTER --------------------------------------------------------------

	public void setEvalStrategy(EvaluationStrategy evalStrategy) {
//		if (evalStrategy == null)
//			throw new IllegalArgumentException("evaluation strategy must not be NULL");
		this.evalStrategy = evalStrategy;
	}
	
	public void setFederationOptimizer(QueryOptimizer optimizer) {
		if (optimizer == null)
			throw new IllegalArgumentException("query optimizer must not be NULL");
		this.optimizer = optimizer;
	}
	
	public void setMembers(List<Repository> members) {
		if (members == null)
			throw new IllegalArgumentException("federation members must not be NULL");
		for (Repository rep : members) {
			if (rep == null)
				throw new IllegalArgumentException("federation member must not be NULL");
		}
		this.members = members;
	}

	public void setSourceSelector(SourceSelector selector) {
		if (selector == null)
			throw new IllegalArgumentException("source selector must not be NULL");
		this.selector = selector;
	}
	
	// -------------------------------------------------------------------------
	
	/**
	 * Initializes the Sail with statistics and optimizer settings.
	 * 
	 * @throws StoreException
	 *         If the Sail could not be initialized.
	 */
	@Override
	public void initialize() throws SailException {
		
		if (initialized) {
			LOGGER.info("Federation Sail is already initialized");
			return;
		}
		
		super.initialize();  // only Sesame 2 needs to initialize super class
		
//		if (this.evalStrategy == null)
//			throw new SailException("Sail evaluation strategy has not been initialized");
		
		// initialize all members
		for (Repository rep : this.members) {
			try {
				rep.initialize();
			} catch (RepositoryException e) {
				throw new SailException("can not initialize repository: " + e.getMessage(), e);
			} catch (IllegalStateException e) {
				LOGGER.debug("member repository is already initialized", e);
			}
		}
		
		// initialize statistics and source selector
		VoidStatistics stats = VoidStatistics.getInstance();
		this.selector.setStatistics(stats);
		this.selector.initialize();
		
		// initialize evaluation strategy
		if (this.evalStrategy == null)
			this.evalStrategy = new FederationEvalStrategy(this.vf);
		
		if (this.evalStrategy instanceof InitializedWithSail) {
			try {
				((InitializedWithSail) this.evalStrategy).init(this);
			} catch (SailConfigException e) {
				throw new SailException("initialization of evaluation strategy failed", e);
			}
		}
		
		initialized = true;
	}
	
	/**
	 * Shuts down the Sail.
	 * 
	 * @throws StoreException
	 *         If the Sail encountered an error or unexpected internal state.
	 */
	@Override
	protected void shutDownInternal() throws SailException {
		for (Repository rep : this.members) {
			try {
				rep.shutDown();
			} catch (RepositoryException e) {
				throw new SailException(e);
			}
		}
	}

	/**
	 * Returns a wrapper for the Sail connections of the federation members.
	 * 
	 * TODO: currently opens connections to *all* federation members,
	 *       should better use lazy initialization of the connections.
	 *       
	 * @return the Sail connection wrapper.
	 */
	@Override
	protected SailConnection getConnectionInternal() throws SailException {
		
		if (!this.initialized)
			throw new IllegalStateException("Sail has not been initialized.");
		
		return new FederationSailConnection(this);
	}

	// SESAME 2 ================================================================
	
	private final ValueFactory vf = new ValueFactoryImpl();
	
	@Override
	public ValueFactory getValueFactory() {
		return vf;
	}

	@Override
	public boolean isWritable() throws SailException {
		return false;
	}
	
} public abstract class AbstractCardinalityEstimator extends QueryModelVisitorBase<RuntimeException> implements ModelEvaluator {

	protected Map<TupleExpr, Double> cardIndex = new HashMap<TupleExpr, Double>();
	
	@Override
	public Double process(TupleExpr expr) {
		synchronized (this) {
			expr.visit(this);
			return cardIndex.get(expr);
		}
	}
	
	@Override
	public void meet(Filter filter) {
		
		// check cardinality index first
		if (getIndexCard(filter) != null)
			return;
		
		// TODO: include condition in estimation
		// for now use same card as sub expression
		filter.getArg().visit(this);
		
		// add same cardinality as filter argument
		setIndexCard(filter, getIndexCard(filter.getArg()));
	}
	
	@Override
	protected void meetUnaryTupleOperator(UnaryTupleOperator node)
			throws RuntimeException {
		if (node instanceof RemoteQuery) {
			meet((RemoteQuery) node);
		} else {
			super.meetUnaryTupleOperator(node);
		}
	}
	
	protected void meet(RemoteQuery node) {
		if (getIndexCard(node) != null)
			return;
		node.getArg().visit(this);
		setIndexCard(node, getIndexCard(node.getArg()));
	}
	
	protected Double getIndexCard(TupleExpr expr) {
		return this.cardIndex.get(expr);
	}
	
	protected void setIndexCard(TupleExpr expr, Double value) {
		this.cardIndex.put(expr, value);
	}

} public abstract class AbstractCostEstimator extends QueryModelVisitorBase<RuntimeException> implements ModelEvaluator {
	
	protected double cost;
	
	protected AbstractCardinalityEstimator cardEst;
	
	public AbstractCardinalityEstimator getCardinalityEstimator() {
		return cardEst;
	}

	public void setCardinalityEstimator(AbstractCardinalityEstimator cardEst) {
		this.cardEst = cardEst;
	}

	public Double getCost(TupleExpr expr) {
		cost = 0;
		expr.visit(this);
		return cost;
	}
	
	@Override
	public Double process(TupleExpr expr) {
		synchronized (this) {
			return getCost(expr);
		}
	}
	
} public class CardinalityCostEstimator extends AbstractCostEstimator {

	@Override
	public String getName() {
		return "CardCost";
	}

	@Override
	protected void meetNode(QueryModelNode node) throws RuntimeException {
		super.meetNode(node);
		if (node instanceof TupleExpr)
			cost += cardEst.process((TupleExpr) node);
	}

} public interface ModelEvaluator {
	
	public Double process(TupleExpr expr);
	
	public String getName();

} public class SPLENDIDCardinalityEstimator extends VoidCardinalityEstimator {
	
	boolean distSOPerPred;
	
	public SPLENDIDCardinalityEstimator(RDFStatistics stats, boolean distSOPerPred) {
		super(stats);
		this.distSOPerPred = distSOPerPred;
	}
	
	@Override
	public String getName() {
		return "SPLDCard";
	}
	
	@Override
	protected Number getPatternCard(MappedStatementPattern pattern, Graph source) {
		
		Value sVal = pattern.getSubjectVar().getValue();
		Value pVal = pattern.getPredicateVar().getValue();
		Value oVal = pattern.getObjectVar().getValue();
		
		// check trivial cave that all variables are bound
		if (sVal != null && pVal != null && oVal != null)
			return 1;
		
		// handle rdf:type
		if (RDF.TYPE.equals(pVal) && oVal != null) {
			return stats.getTypeCount(source, oVal.stringValue());
		}
		
		Number resultSize;
		if (pVal == null) {
			resultSize = stats.getTripleCount(source);
		} else {
			resultSize = stats.getPredicateCount(source, pVal.stringValue());
		}
		
		// object is bound
		if (oVal != null) {
			if (distSOPerPred && pVal != null) {
				long distPredObj = stats.getDistinctObjects(source, pVal.stringValue());
				if (distPredObj == -1)
					throw new IllegalArgumentException("no value for distinct Objects per Predicate in statistics");
				return resultSize.doubleValue() / distPredObj;
			} else {
				long pCount = stats.getDistinctPredicates(source);
				long distObj = stats.getDistinctObjects(source);
				return resultSize.doubleValue() * pCount / distObj; 
			}
		}
		
		// subject is bound
		if (sVal != null) {
			if (distSOPerPred && pVal != null) {
				long distPredSubj = stats.getDistinctSubjects(source, pVal.stringValue());
				if (distPredSubj == -1)
					throw new IllegalArgumentException("no value for distinct Objects per Predicate in statistics");
				return resultSize.doubleValue() / distPredSubj;
			} else {
				long pCount = stats.getDistinctPredicates(source);
				long distSubj = stats.getDistinctSubjects(source);
				return resultSize.doubleValue() * pCount / distSubj; 
			}
			
		}

		// use triple count containing the predicate
		return resultSize;
	}

} public class SPLENDIDCostEstimator extends AbstractCostEstimator {
	
	private static final Logger LOGGER = LoggerFactory.getLogger(SPLENDIDCostEstimator.class);
	
	private static final int C_TRANSFER_QUERY = 5;
	private static final int C_TRANSFER_TUPLE = 1;
	
	public String getName() {
		return "SPLDCost";
	}
	
	@Override
	public void meet(Join node) throws RuntimeException {
		
		// get cost of children first
		super.meet(node);
		
		if (node instanceof HashJoin) {
			meet((HashJoin) node);
		} else if (node instanceof BindJoin) {
			meet((BindJoin) node);
		} else {
			throw new IllegalArgumentException("no accepted join: " + node);
		}
	}

	@Override
	protected void meetUnaryTupleOperator(UnaryTupleOperator node)
			throws RuntimeException {
		
		// need to handle sub queries explicitly
		if (node instanceof RemoteQuery) {
			meet((RemoteQuery) node);
		} else {
			super.meetUnaryTupleOperator(node);
		}
	}
	
	// -------------------------------------------------------------------------
	
	protected void meet(HashJoin join) {
		Double leftCard = cardEst.process(join.getLeftArg());
		Double rightCard = cardEst.process(join.getRightArg());
		
		this.cost += (leftCard + rightCard) * C_TRANSFER_TUPLE + 2 * C_TRANSFER_QUERY;
		
//		LOGGER.warn("HashJoin: " + leftCard + " >< " + rightCard + " :: " + (leftCard + rightCard) * C_TRANSFER_TUPLE + 2 * C_TRANSFER_QUERY);
	}
	
	protected void meet(BindJoin join) {
		Double leftCard = cardEst.process(join.getLeftArg());
		Double joinCard = cardEst.process(join);
		
		this.cost += leftCard * (C_TRANSFER_TUPLE + C_TRANSFER_QUERY) + joinCard * C_TRANSFER_TUPLE;
		
//		LOGGER.warn("BindJoin: " + leftCard + " >< " + joinCard + " :: " + (leftCard + joinCard) * C_TRANSFER_TUPLE + 2 * C_TRANSFER_QUERY);
	}
	
	protected void meet(RemoteQuery query) {
		
		// no explicit cost for executing the sub query
		
	}

} public class TrueCardinalityEstimator extends AbstractCardinalityEstimator {

	private static final Logger LOGGER = LoggerFactory.getLogger(TrueCardinalityEstimator.class);
	
	private EvaluationStrategy evalStrategy;
	
	public TrueCardinalityEstimator(EvaluationStrategy evalStrategy) {
		this.evalStrategy = evalStrategy;
	}
	
	@Override
	public String getName() {
		return "TrueCard";
	}
	
	@Override
	protected void meetNode(QueryModelNode node) throws RuntimeException {
		if (node instanceof TupleExpr) {
			TupleExpr expr = (TupleExpr) node;
			
			// check cardinality index first
			if (getIndexCard(expr) != null)
				return;
			
			try {
				int card = QueryExecutor.getSize(evalStrategy.evaluate(expr, EmptyBindingSet.getInstance()));
				
				// add cardinality to index
				setIndexCard(expr, (double) card);
				
			} catch (QueryEvaluationException e) {
				throw new RuntimeException("query evaluation failed", e);
			}
			
		} else {
			throw new IllegalArgumentException();
		}
	}
} public abstract class VoidCardinalityEstimator extends AbstractCardinalityEstimator {
	
	private static final Logger LOGGER = LoggerFactory.getLogger(VoidCardinalityEstimator.class);
	
	protected RDFStatistics stats;
	
	// -------------------------------------------------------------------------
	
	protected abstract Number getPatternCard(MappedStatementPattern pattern, Graph source);
	
	// -------------------------------------------------------------------------
	
	public VoidCardinalityEstimator(RDFStatistics stats) {
		if (stats == null)
			throw new IllegalArgumentException("RDF stats must not be NULL.");
		
		this.stats = stats;
	}
	
	public void meet(MappedStatementPattern pattern) throws RuntimeException {
		
		// check cardinality index first
		if (getIndexCard(pattern) != null)
			return;
		
		double card = 0;

		// estimate the cardinality of the pattern for each source and sum up
		// assumes that all result tuple will be distinct
		for (Graph source : pattern.getSources()) {
			card += getPatternCard(pattern, source).doubleValue();
		}
		
		// add cardinality to index
		setIndexCard(pattern, card);
	}
	
	/**
	 * Group a list of triple patterns by subject.
	 * 
	 * @param patterns the patterns to group.
	 * @return a mapping of subject variables to triple patterns.
	 */
	private Map<Var, List<StatementPattern>> groupBySubject(List<StatementPattern> patterns) {
		Map<Var, List<StatementPattern>> patternGroups = new HashMap<Var, List<StatementPattern>>();
		
		for (StatementPattern p : patterns) {
			
			Var subjectVar = p.getSubjectVar();
			List<StatementPattern> pList = patternGroups.get(subjectVar);
			if (pList == null) {
				pList = new ArrayList<StatementPattern>();
				patternGroups.put(subjectVar, pList);
			}
			pList.add(p);
		}
		return patternGroups;
	}
	
	/**
	 * Returns the cardinality of the supplied pattern.
	 * 
	 * @param p the pattern.
	 * @return the cardinality of the pattern.
	 */
	private double getCardinality(StatementPattern p) {
		Double card = getIndexCard(p);
		if (card == null) {
			meet(p);
			card = getIndexCard(p);
		}
		return card;
	}

	/**
	 * Returns the get selectivity of the supplied pattern.
	 * Based on the distinct subjects per source.
	 * 
	 * @param p the pattern.
	 * @return the selectivity of the pattern.
	 */
	private double getSelectivity(StatementPattern p) {
		long distinctSubjects = 0;
		Set<Graph> sources = ((MappedStatementPattern) p).getSources();
		Value pValue = p.getPredicateVar().getValue();
		
		for (Graph source : sources) {
			// TODO: check that predicate value is not null
			if (pValue == null) {
				distinctSubjects += stats.getDistinctSubjects(source);
			} else {
				distinctSubjects += stats.getDistinctSubjects(source, pValue.stringValue());
			}
		}
		return 1.0 / distinctSubjects;
	}
	
	private double getJoinSelectivity(String varName, List<StatementPattern> leftPatterns, List<StatementPattern> rightPatterns) {

		// compute minimum of all selectivity values (left or right)
		double selectivity = 1;

		// get left pattern selectivity (minimum of all selectivity values)
		for (StatementPattern p : leftPatterns) {
			if (VarNameCollector.process(p).contains(varName)) {
				// get left pattern selectivity
				double pSel = getVarSelectivity((MappedStatementPattern) p, varName);
				if (Double.compare(pSel, selectivity) < 0)
					selectivity = pSel;
			}
		}
		
		// get right pattern selectivity (minimum of all selectivity values)
		for (StatementPattern p : leftPatterns) {
			if (VarNameCollector.process(p).contains(varName)) {
				double pSel = getVarSelectivity((MappedStatementPattern) p, varName);
				if (Double.compare(pSel, selectivity) < 0)
					selectivity = pSel;
			}
		}
		
		return selectivity;
	}
	
	private double computeSubjectBasedCardinality(List<StatementPattern> patterns, Map<Var, Double> sVarCardinality, Map<Var, Double> sVarSelectivity) {
		
		Map<Var, List<StatementPattern>> sVarGroups = groupBySubject(patterns);
		for (Var var : sVarGroups.keySet()) {
			
			List<StatementPattern> sVarPatterns = sVarGroups.get(var);
			
			// handle bound subject variables
			if (var.getValue() != null) {
				for (StatementPattern p : sVarPatterns) {
					double cardinality = getCardinality(p);
					double selectivity = getSelectivity(p);
					sVarCardinality.put(var, cardinality);
					sVarSelectivity.put(var, selectivity);
				}
				continue;
			}
			
			// TODO: optimization: handle list with only one element
			
			boolean boundPO = false;
			Double minPOCardinality = Double.POSITIVE_INFINITY;
			Double minPOSelectivity = Double.POSITIVE_INFINITY;
			List<Double> nonPOCardinalities = new ArrayList<Double>();
			List<Double> nonPOSelectivities = new ArrayList<Double>();
			
			// process all pattern with the same subject variable
			// and compute the join cardinality + selectivity
			for (StatementPattern p : sVarPatterns) {
				
				double cardinality = getCardinality(p);
				double selectivity = getSelectivity(p);
				
				// compute minimum cardinality/selectivity only if P+O are bound
				if (p.getPredicateVar().getValue() != null && p.getObjectVar().getValue() != null) {
					if (Double.compare(cardinality, minPOCardinality) < 0)
						minPOCardinality = cardinality;
					if (Double.compare(selectivity, minPOSelectivity) < 0)
						minPOSelectivity = selectivity;
					boundPO = true;
				} else {
					nonPOCardinalities.add(cardinality);
					nonPOSelectivities.add(selectivity);
				}
			}
			
			// add minimum cardinality/selectivity if computed for patterns with bound P+O
			if (boundPO) {
				nonPOCardinalities.add(minPOCardinality);
				nonPOSelectivities.add(minPOSelectivity);
			}
			
			// compute the join cardinality over all patterns with the same subject 
			Collections.sort(nonPOSelectivities);
			double joinCardinality = nonPOCardinalities.get(0);
			for (int i = 1; i < nonPOCardinalities.size(); i++) {
				joinCardinality *= nonPOCardinalities.get(i);
				joinCardinality *= nonPOSelectivities.get(i-1);
			}
			
			// TODO: check for a possible second join variable in the patterns 
			
			// store cardinality for set of patterns with same subject variable
			// store minimum selectivity 
			sVarCardinality.put(var, joinCardinality);
			sVarSelectivity.put(var, nonPOSelectivities.get(0));

			if (LOGGER.isDebugEnabled()) {
				LOGGER.debug("VAR ?" + var.getName() + " [" + sVarPatterns.size() + "]: " + sVarPatterns);
				LOGGER.debug("VAR ?" + var.getName() + ": card=" + joinCardinality + ", sel=" + nonPOSelectivities.get(0));
			}
		}
		
		// collect join variables for each group
		Map<Var, Set<String>> joinVarMap = new HashMap<Var, Set<String>>();
		
		for (Var var : sVarGroups.keySet()) {
			List<StatementPattern> sVarPatterns = sVarGroups.get(var);
			
			// collect variable names
			Set<String> vars = new HashSet<String>();
			for (StatementPattern p : sVarPatterns) {
				vars.addAll(VarNameCollector.process(p));
			}
			joinVarMap.put(var, vars);
		}
		
		// compute join selectivity for all possible combination of the groups
		List<Double> selectivityList = new ArrayList<Double>();
		List<Var> vars = new ArrayList<Var>(joinVarMap.keySet());
		for (int i = 0; i < vars.size(); i++) {
			for (int j = i+1; j < vars.size(); j++) {
				Set<String> joinVars = new HashSet<String>(joinVarMap.get(vars.get(i)));
				joinVars.retainAll(joinVarMap.get(vars.get(j)));
				
				double selectivity = 1;
				// if not a cross product: selectivity < 1
				if (!joinVars.isEmpty()) {
					// get selectivity for both join arguments
					
					if (joinVars.size() > 1)
						LOGGER.warn("join estimation for multiple vars");
					
					List<StatementPattern> leftPatterns = sVarGroups.get(vars.get(i));
					List<StatementPattern> rightPatterns = sVarGroups.get(vars.get(j));
					
					// compute selectivity for all join variables (usually just one) and take minimum
					for (String varName : joinVars) {
//						LOGGER.warn("JS " + varName + ": " + leftPatterns + " <-> " + rightPatterns);
						
						double jSel = getJoinSelectivity(varName, leftPatterns, rightPatterns);
						if (Double.compare(jSel, selectivity) < 0)
							selectivity = jSel;
					}
				}
				// add selectivity to list of selectivity values
				selectivityList.add(selectivity);
				
//				LOGGER.warn("compare: " + selectivity + " for " + joinVars + " <-- " + joinVarMap.get(vars.get(i)) + " <-> " + joinVarMap.get(vars.get(j)));
			}
		}
		
		// sort selectivity list
		Collections.sort(selectivityList);
		
		double joinCardinality = sVarCardinality.get(vars.get(0));
		for (int i = 1; i < vars.size(); i++) {
			joinCardinality *= sVarCardinality.get(vars.get(i));
			joinCardinality *= sVarSelectivity.get(vars.get(i-1));
		}
		
//		LOGGER.warn("join cardinality: " + joinCardinality);
		
		return joinCardinality;
	}
	
	// -------------------------------------------------------------------------
	
	@Override
	public void meet(StatementPattern pattern) throws RuntimeException {
		if (pattern instanceof MappedStatementPattern)
			meet((MappedStatementPattern) pattern);
		else
			throw new IllegalArgumentException("cannot estimate cardinality for triple pattern without sources: " + pattern);
	}
	
	/**
	 * Computing the cardinality of a join.
	 * It does not depend on the join order of the triple patterns.
	 * First, joins with the same subject variable are specially treated.
	 * Then, object-object joins and subject-objects joins are handled.
	 * 
	 * @param join the joins used for the cardinality computation.
	 */
	@Override
	public void meet(Join join) throws RuntimeException {
		
		// check cardinality index first
		if (getIndexCard(join) != null)
			return;
		
		// collect all statement patterns
		List<StatementPattern> patterns = StatementPatternCollector.process(join);
		
		// ----------------------------------------------------------
		
		Map<Var, Double> sVarCardinality = new HashMap<Var, Double>();
		Map<Var, Double> sVarSelectivity = new HashMap<Var, Double>();
		
		// group by subject variable and compute cardinality/selectivity for each group
		double card = computeSubjectBasedCardinality(patterns, sVarCardinality, sVarSelectivity);
		
		// add cardinality to index
		setIndexCard(join, card);
		
//		// ----------------------------------------------------------
//		
//		// TODO: does the estimated cardinality depend on the current join?
//		//       -> no: different join order should yield same cardinality
//		
//		// estimate cardinality of join arguments first
//		join.getLeftArg().visit(this);
//		join.getRightArg().visit(this);
//		
//		double joinSelectivity = getJoinSelectivity(join.getLeftArg(), join.getRightArg());
//		
//		double leftCard = getIndexCard(join.getLeftArg());
//		double rightCard = getIndexCard(join.getRightArg());
//		double card = joinSelectivity * leftCard * rightCard;
//
//		// add cardinality to index
//		setIndexCard(join, card);
	}
	
	// -------------------------------------------------------------------------
	
	/**
	 * Get the selectivity for a (named) variable in the supplied pattern.
	 * Computes the selectivity as 1/sum(card_i(P)) for all data sources 'i'. 
	 * 
	 * @param pattern the query pattern to process.
	 * @param varName the variable name.
	 * @return the selectivity of the variable.
	 */
	protected double getVarSelectivity(MappedStatementPattern pattern, String varName) {
		long count = 0;
		
		// TODO: this does not look right yet
		
		for (Graph source : pattern.getSources()) {
			
			Long pCount = stats.getDistinctPredicates(source);
			
			if (varName.equals(pattern.getSubjectVar().getName())) {
				count += stats.getDistinctSubjects(source);
//				count += stats.distinctSubjects(source) / pCount.doubleValue();
				continue;
			}
			if (varName.equals(pattern.getPredicateVar().getName())) {
				throw new UnsupportedOperationException("predicate join not supported yet");
			}
			if (varName.equals(pattern.getObjectVar().getName())) {
				count += stats.getDistinctObjects(source);
//				count += stats.distinctObjects(source) / pCount.doubleValue();
				continue;
			}
			throw new IllegalArgumentException("var name not found in pattern");
		}
		return 1.0 / count;
	}
	
	protected double getJoinSelectivity(TupleExpr leftExpr, TupleExpr rightExpr) {
		
		// get join variables
		Set<String> joinVars = VarNameCollector.process(leftExpr);
		joinVars.retainAll(VarNameCollector.process(rightExpr));
		
		if (joinVars.size() == 0) {
			// cross product: selectivity is 1
			return 1.0;
		}
		if (joinVars.size() == 2) {
			// multi-valued join
			LOGGER.warn("join estimation for multiple vars not supported (yet) - using first var only");
		}
		
		// get all patterns which contain the join variables
		// TODO: extend to more than one join variable
		String joinVar = joinVars.iterator().next();
		List<StatementPattern> leftJoinPatterns = new ArrayList<StatementPattern>();
		List<StatementPattern> rightJoinPatterns = new ArrayList<StatementPattern>();
		
		for (StatementPattern pattern : StatementPatternCollector.process(leftExpr)) {
			if (VarNameCollector.process(pattern).contains(joinVar))
				leftJoinPatterns.add(pattern);
		}
		for (StatementPattern pattern : StatementPatternCollector.process(rightExpr)) {
			if (VarNameCollector.process(pattern).contains(joinVar))
				rightJoinPatterns.add(pattern);
		}
		
		// select one pattern from each join argument to define the join condition
		// TODO: analyze structure of join, current approach produces random estimations
		double leftSel = getVarSelectivity((MappedStatementPattern) leftJoinPatterns.get(0), joinVar);
		double rightSel = getVarSelectivity((MappedStatementPattern) rightJoinPatterns.get(0), joinVar);
		return Math.min(leftSel, rightSel);
	}
	
} public class SubQueryBuilder {
	
	private static final Logger LOGGER = LoggerFactory.getLogger(SubQueryBuilder.class);
	
	private boolean groupBySameAs;
	private boolean groupBySource;
	
	public SubQueryBuilder(QueryOptimizerConfig config) {
		this.groupBySource = config.isGroupBySource();
		this.groupBySameAs = config.isGroupBySameAs();
	}

	/**
	 * Creates sub queries for all mapped data sources.
	 * 
	 * @param patterns the statement patterns with mapped data sources.
	 * @param conditions the filter conditions to apply.
	 * @return a list of created sub queries.
	 */
	public List<TupleExpr> createSubQueries(List<MappedStatementPattern> patterns, List<ValueExpr> conditions) {
		
		if (patterns == null || patterns.size() == 0)
			throw new IllegalArgumentException("need at least on triple pattern to create sub queries");
		
		List<TupleExpr> subQueries = new ArrayList<TupleExpr>();
		
		// create groups of triple patterns according to the configuration
		List<List<MappedStatementPattern>> groups = getGroups(patterns);
		
		// create remote queries for all triple pattern groups
		// triple patterns in groups should all have the same sources
		for (List<MappedStatementPattern> patternGroup : groups) {
			
			// check that all patterns in a group have the same set of source
			// TODO: needs improvement - not the best way to do it
			Set<Graph> sources = null;
			for (MappedStatementPattern pattern : patternGroup) {
				if (sources == null)
					sources = pattern.getSources();
				else
					if (!sources.equals(pattern.getSources())) {
						LOGGER.warn("Statement Patterns with different sources in group");
						sources.addAll(pattern.getSources());
					}
			}
			
			TupleExpr baseExpr = null;
			
			// create a remote query if the pattern group has a single source
			if (sources.size() == 1) {
				for (MappedStatementPattern pattern : patternGroup) {
					baseExpr = (baseExpr == null) ? pattern : new Join(baseExpr, pattern);
				}
				baseExpr = applyFilters(baseExpr, conditions);
				subQueries.add(new RemoteQuery(baseExpr));
			}
			
			// create individual remote queries if there is more than one source 
			else {
				for (MappedStatementPattern pattern : patternGroup) {
					baseExpr = applyFilters(pattern, conditions);
					subQueries.add(new RemoteQuery(baseExpr));
				}
			}
		}
		
		return subQueries;
	}
	
	private TupleExpr applyFilters(TupleExpr expr, List<ValueExpr> conditions) {
		Set<String> varNames = VarNameCollector.process(expr);
		for (ValueExpr condition : conditions) {
			if (varNames.containsAll(VarNameCollector.process(condition))) {
				expr = new Filter(expr, condition);
			}
		}
		return expr;
	}
	
	public List<List<MappedStatementPattern>> getGroups(List<MappedStatementPattern> patterns) {
		
		// sameAs grouping example:
		// ?city :population ?x  -> [A]
		// ?city :founded_in ?y  -> [B]
		// ?city owl:sameAs  ?z  -> [A,C,D]  ... group with first pattern only
		
		// 1. no grouping -> put each single pattern in a set
		// 2. group by source -> put pattern in set based on same source
		// 3. group by sameAs -> put sameAs pattern in set with matched pattern
		// 4. group by source and by sameAs
		
		List<MappedStatementPattern> sameAsPatterns = new ArrayList<MappedStatementPattern>();
		List<List<MappedStatementPattern>> patternGroups = new ArrayList<List<MappedStatementPattern>>();
		
		// start with grouping of sameAs patterns if applicable
		// this removes all sameAs patterns and matched pattern from the list
		if (groupBySameAs) {
			
			// move sameAs patterns from pattern list to sameAs pattern list
			Iterator<MappedStatementPattern> it = patterns.iterator();
			while (it.hasNext()) {
				MappedStatementPattern pattern = it.next();
				if (OWL.SAMEAS.equals(pattern.getPredicateVar().getValue()) && !pattern.getSubjectVar().hasValue()) {
					sameAsPatterns.add(pattern);
					it.remove();
				}
			}
			
			Set<MappedStatementPattern> matchedPatterns = new HashSet<MappedStatementPattern>();
			
			// find all matching pattern for each sameAs pattern
			for (MappedStatementPattern sameAsPattern : sameAsPatterns) {
				
				List<MappedStatementPattern> matchCandidates = new ArrayList<MappedStatementPattern>();
				Set<Graph> sameAsSources = sameAsPattern.getSources();
				Var sameAsSubjectVar = sameAsPattern.getSubjectVar();
				
				// find match candidates
				for (MappedStatementPattern pattern : patterns) {
					Set<Graph> patternSources = pattern.getSources();
					// check if pattern meets condition
					if (containsVar(pattern, sameAsSubjectVar) && sameAsSources.containsAll(patternSources)) {
						matchCandidates.add(pattern);
					}
				}
				
				// check if any match candidate was found
				if (matchCandidates.size() == 0) {
					// found no patterns to match with sameAs
					// add sameAs pattern as its own group
					List<MappedStatementPattern> group = new ArrayList<MappedStatementPattern>();
					group.add(sameAsPattern);
					patternGroups.add(group);
					continue;
				}
				
				// group match candidates with sameAs pattern
				for (MappedStatementPattern pattern : matchCandidates) {
					matchedPatterns.add(pattern);
					// add sameAs pattern with the matched pattern's Sources
					List<MappedStatementPattern> group = new ArrayList<MappedStatementPattern>();
					group.add(pattern);
					group.add(new MappedStatementPattern(sameAsPattern, pattern.getSources()));
					patternGroups.add(group);
				}
			}
			// removed all matched patterns from the pattern list
			patterns.removeAll(matchedPatterns);
		}
		
		if (groupBySource) {
			
			// create map for {Source}->{Pattern}
			Map<Set<Graph>, List<MappedStatementPattern>> sourceMap = new HashMap<Set<Graph>, List<MappedStatementPattern>>();
			
			// add all sameAs groups first, if applicable
			if (groupBySameAs) {
				for (List<MappedStatementPattern> patternList : patternGroups) {
					Set<Graph> sources = patternList.get(0).getSources();
					List<MappedStatementPattern> pList = sourceMap.get(sources);
					if (pList == null) {
						pList = new ArrayList<MappedStatementPattern>();
						sourceMap.put(sources, pList);
					}
					pList.addAll(patternList);
				}
			}
			
			// add all pattern from the list to the source mapping
			for (MappedStatementPattern pattern : patterns) {
				Set<Graph> sources = pattern.getSources();
				List<MappedStatementPattern> pList = sourceMap.get(sources);
				if (pList == null) {
					pList = new ArrayList<MappedStatementPattern>();
					sourceMap.put(sources, pList);
				}
				pList.add(pattern);
			}
			
			// finally create pattern groups for all source mappings
			patternGroups.clear();
//			patternGroups.addAll(sourceMap.values());
			
			// keep groups only if there is no more than one source assigned
			for (Set<Graph> graphs : sourceMap.keySet()) {
				List<MappedStatementPattern> pGroup = sourceMap.get(graphs);
				if (graphs.size() == 1) {
					patternGroups.add(pGroup);
				} else {
					for (MappedStatementPattern pattern : pGroup) {
						List<MappedStatementPattern> pList = new ArrayList<MappedStatementPattern>();
						pList.add(pattern);
						patternGroups.add(pList);
					}
				}
			}
			
		} else {
			
			// add all patterns from the list as individual group
			for (MappedStatementPattern pattern : patterns) {
				List<MappedStatementPattern> pList = new ArrayList<MappedStatementPattern>();
				pList.add(pattern);
				patternGroups.add(pList);
			}
			
		}
		
		// debugging
		for (List<MappedStatementPattern> pList : patternGroups) {
			StringBuffer buffer = new StringBuffer("Group [");
			Set<Graph> sources = null;
			for (MappedStatementPattern pattern : pList) {
				buffer.append(OperatorTreePrinter.print(pattern)).append(", ");
				if (sources == null) {
					sources = pattern.getSources();
				} else {
					if (!sources.equals(pattern.getSources()))
						LOGGER.warn("not the same sources: " + sources + " <-> " + pattern.getSources());
				}
			}
			buffer.setLength(buffer.length()-2);
			buffer.append("] @" + sources);
			
			if (LOGGER.isDebugEnabled())
				LOGGER.debug(buffer.toString());
		}
		
		return patternGroups;
	}

	private boolean containsVar(StatementPattern pattern, Var var) {
		String varName = var.getName();
		Var sVar = pattern.getSubjectVar();
		Var oVar = pattern.getObjectVar();
		return (!sVar.hasValue() && sVar.getName().equals(varName))
			|| (!oVar.hasValue() && oVar.getName().equals(varName));
	}

} public abstract class AbstractFederationOptimizer implements QueryOptimizer {
	
	private static final Logger LOGGER = LoggerFactory.getLogger(AbstractFederationOptimizer.class);
	
	protected SourceSelector sourceSelector;
	protected SubQueryBuilder queryBuilder;
	protected AbstractCostEstimator costEstimator;
	protected ModelEvaluator modelEvaluator;
	
	/**
	 * To be implemented by sub classes.
	 * 
	 * @param query the Query to optimize.
	 */
	public abstract TupleExpr optimizeBGP(TupleExpr query);
	
	// -------------------------------------------------------------------------
	
	public SourceSelector getSelector() {
		return sourceSelector;
	}

	public void setSelector(SourceSelector sourceSelector) {
		if (sourceSelector == null)
			throw new IllegalArgumentException("source selector must not be null");
		this.sourceSelector = sourceSelector;
	}

	public SubQueryBuilder getBuilder() {
		return queryBuilder;
	}

	public void setBuilder(SubQueryBuilder queryBuilder) {
		if (queryBuilder == null)
			throw new IllegalArgumentException("pattern group builder must not be null");
		this.queryBuilder = queryBuilder;
	}

	public AbstractCostEstimator getCostEstimator() {
		return costEstimator;
	}

	public void setCostEstimator(AbstractCostEstimator costEstimator) {
		if (costEstimator == null)
			throw new IllegalArgumentException("cost estimator must not be null");
		this.costEstimator = costEstimator;
	}
	
	public ModelEvaluator getModelEvaluator() {
		return modelEvaluator;
	}

	public void setModelEvaluator(ModelEvaluator modelEvaluator) {
		this.modelEvaluator = modelEvaluator;
	}
	
	// -------------------------------------------------------------------------
	
	protected List<TupleExpr> getBaseExpressions(TupleExpr expr) {
		
		// get patterns and filter conditions from query model
		List<StatementPattern> patterns = StatementPatternCollector.process(expr);
		List<ValueExpr> conditions = FilterConditionCollector.process(expr);
		
		// create patterns with source mappings
		List<MappedStatementPattern> mappedPatterns = this.sourceSelector.mapSources(patterns);
		
		return this.queryBuilder.createSubQueries(mappedPatterns, conditions);
	}
	
	@Override
	public void optimize(TupleExpr query, Dataset dataset, BindingSet bindings) {  // Sesame 2
		
		// collect all basic graph patterns
		for (TupleExpr bgp : BasicGraphPatternExtractor.process(query)) {
			
//			// a single statement pattern needs no optimization
//			// TODO: need sources first
//			if (bgp instanceof StatementPattern)
//				continue;
			
			if (LOGGER.isTraceEnabled())
				LOGGER.trace("BGP before optimization:\n" + AnnotatingTreePrinter.print(bgp));

			bgp = optimizeBGP(bgp);
			
			if (LOGGER.isTraceEnabled() && modelEvaluator != null)
				LOGGER.trace("BGP after optimization:\n" + AnnotatingTreePrinter.print(bgp, modelEvaluator));
		}
		
	}
	
} public class DynamicProgrammingOptimizer extends AbstractFederationOptimizer {
	
	private static final Logger LOGGER = LoggerFactory.getLogger(DynamicProgrammingOptimizer.class);
	
	private boolean bindJoin;
	private boolean hashJoin;
			
	public DynamicProgrammingOptimizer(boolean hashJoin, boolean bindJoin) {
		if (hashJoin == false && bindJoin == false)
			throw new IllegalArgumentException("cannot create joins: all physical join types are disabled");
		this.bindJoin = bindJoin;
		this.hashJoin = hashJoin;
	}

	@Override
	public TupleExpr optimizeBGP(TupleExpr bgp) {
		
		long time = System.currentTimeMillis();

		PlanCollection optPlans = new PlanCollection();
		List<ValueExpr> conditions = FilterConditionCollector.process(bgp);

		// create access plans for all statement patterns
		List<TupleExpr> plans = this.getBaseExpressions(bgp);
		int count = plans.size();
		prune(plans);
		optPlans.add(new HashSet<TupleExpr>(plans), 1);
		
		// create all n-ary join combinations [n = 2 .. #plans]
		for (int n = 2; n <= count; n++) {
			
			// build n-ary joins by combining i-ary joins and (n-i)-ary joins
			for (int i = 1; i < n; i++) {
				
				Set<TupleExpr> plans1 = optPlans.get(i);
				Set<TupleExpr> plans2 = optPlans.get(n-i);
				
				// find for each plan all complementary plans to join
				for (TupleExpr plan : plans1) {
					Set<TupleExpr> comp = filterDistinctPlans(plan, plans2);
					// we may not always find complementary plans
					if (comp.size() > 0) {
						// cross products are created if there is no common join variable
						plans = createJoins(plan, comp, conditions);
						optPlans.add(new HashSet<TupleExpr>(plans), n);
					}
				}
			}
			
			if (LOGGER.isTraceEnabled())
				LOGGER.trace(optPlans.get(n).size() + " plans generated for N=" + n);
			
			prune(optPlans.get(n));
		}
		
		if (LOGGER.isDebugEnabled())
			LOGGER.debug("time taken for optimization: " + (System.currentTimeMillis() - time));
		
		// return the final plan
		if (optPlans.get(count).size() == 0)
			// This is wrong: how do we get here?
			throw new UnsupportedOperationException("Queries requiring cross products are not supported (yet)");
		
		TupleExpr newOp = optPlans.get(count).iterator().next();
		bgp.replaceWith(newOp);
		return newOp;

	}
	
	/**
	 * Combine tuple expressions with a join operator.
	 * 
	 * @param leftArg the left join argument.
	 * @param rightArgs a set of potential right join arguments.
	 * @param filters filter expressions which may be applied.
	 * @return a list of created physical join operators.
	 */
	public List<TupleExpr> createJoins(TupleExpr leftArg, Set<TupleExpr> rightArgs, List<ValueExpr> filters) {
		
		List<TupleExpr> newPlans = new ArrayList<TupleExpr>();
		
		// check each pair of potential join arguments if they have at leas one variable in common 
		for (TupleExpr plan : rightArgs) {
			
			// get the intersection of variables found in the left and right join argument
			Set<String> vars = VarNameCollector.process(leftArg);
			vars.retainAll(VarNameCollector.process(plan));
			
			// avoid joins with cross products (without a common join variable)
			if (vars.size() == 0) continue;
			
			// create all physical join variations
			for (TupleExpr join : createPhysicalJoins(leftArg, plan)) {
				join = applyFilters(join, filters);
				newPlans.add(join);
			}
		}
		
		// only cross products possible
		if (newPlans.size() == 0) {
			for (TupleExpr plan : rightArgs) {
				// create all physical join variations
				for (TupleExpr join : createPhysicalJoins(leftArg, plan)) {
					join = applyFilters(join, filters);
					newPlans.add(join);
				}
			}
		}
		
		return newPlans;
	}
	
	private List<Join> createPhysicalJoins(TupleExpr leftArg, TupleExpr rightArg) {
		List<Join> joins = new ArrayList<Join>();
		if (bindJoin) {
			// need to clone the argument to maintain the proper parent reference
			TupleExpr left = leftArg.clone();
			TupleExpr right = rightArg.clone();
			Join join = new BindJoin(left, right);
			left.setParentNode(join);
			right.setParentNode(join);
			joins.add(join);
		}
		if (hashJoin) {
			// need to clone the argument to maintain the proper parent reference
			TupleExpr left = leftArg.clone();
			TupleExpr right = rightArg.clone();
			Join join = new HashJoin(left, right);
			left.setParentNode(join);
			right.setParentNode(join);
			joins.add(join);
		}
		return joins;
	}
	
	protected TupleExpr applyFilters(TupleExpr operator, List<ValueExpr> conditions) {
		
		// copy given filters and remove all which are already applied
		Set<ValueExpr> newFilters = new HashSet<ValueExpr>(conditions);
		newFilters.removeAll(FilterConditionCollector.process(operator));
		
		
		// check if node satisfies all variables for remaining filters
		Set<String> nodeVars = VarNameCollector.process(operator);
		Iterator<ValueExpr> it = newFilters.iterator();
		while (it.hasNext()) {
			ValueExpr filter = it.next();
			if (!nodeVars.containsAll(VarNameCollector.process(filter)))
				it.remove();
		}
		
		// apply filters
		for (ValueExpr condition : newFilters) {
			operator = new Filter(operator, condition);
		}
		
		return operator;
	}
	
	public Set<Set<TupleExpr>> groupEquivalentPlans(Collection<TupleExpr> plans) {
		
		// put query plans in same category if the have the same patterns
		Map<Set<StatementPattern>, Set<TupleExpr>> equivalent = new HashMap<Set<StatementPattern>, Set<TupleExpr>>();

		// Collect all equivalent plans, i.e. plans with the same patterns
		for (TupleExpr plan : plans) {
			Set<StatementPattern> patterns = new HashSet<StatementPattern>(StatementPatternCollector.process(plan));
			Set<TupleExpr> planSet = equivalent.get(patterns);
			
			if (planSet == null) {
				planSet = new HashSet<TupleExpr>();
				equivalent.put(patterns, planSet);
			}
			
			planSet.add(plan);
		}
		return new HashSet<Set<TupleExpr>>(equivalent.values());
	}
	
	public Set<TupleExpr> filterDistinctPlans(TupleExpr plan, Set<TupleExpr> plans) {
		
		// distinct plans don't share any triple patterns (intersection is empty)
		
		Collection<StatementPattern> planpatterns = StatementPatternCollector.process(plan);
		
		Set<TupleExpr> planSet = new HashSet<TupleExpr>();
		for (TupleExpr other : plans) {
			Collection<StatementPattern> patterns = StatementPatternCollector.process(other);
			patterns.retainAll(planpatterns);
			if (patterns.size() == 0)
				planSet.add(other);
		}
		return planSet;
	}
	
	/**
	 * Prune inferior plans - only keep plans with the lowest cost estimate.
	 * 
	 * Computes the cost for each plan and retains only the best plans.
	 * Since some plans are not comparable there can be several best plans.
	 */
	private void prune(Collection<TupleExpr> plans) {

		int eqClass = 0;
		Set<TupleExpr> bestPlans = new HashSet<TupleExpr>();
		Set<Set<TupleExpr>> equivalentPlans = groupEquivalentPlans(plans);

//		// DEBUG
//		if (LOGGER.isDebugEnabled()) {
//			String arity = DEBUG_N < 2 ? "access" : DEBUG_N + "-ary join";
//			LOGGER.debug("pruning " + plans.size() + " " + arity + " plans in " + equivalentPlans.size() + " equivalence classes");
//			if (LOGGER.isTraceEnabled()) {
//				int eqs = 0;
//				for (Set<O> equalSet : equivalentPlans) {
//					eqs++;
//					for (O plan : equalSet) {
//						String pStr = plan.toString().replace("\n", " ");
//						LOGGER.trace("SET [" + eqs + "] " + pStr);
//					}
//				}
//			}
//		}
		
		// Compare all plan from same equivalence class to find best one
		for (Set<TupleExpr> equalSet : equivalentPlans) {
			
			int planCount = 0;
			eqClass++;

			TupleExpr bestPlan = null;
			double lowestCost = Double.MAX_VALUE;
			
			for (TupleExpr plan : equalSet) {
				
				planCount++;
				double cost = costEstimator.process(plan);
				
				if (bestPlan == null || cost < lowestCost) {
					
					if (LOGGER.isTraceEnabled() && bestPlan != null) {
						LOGGER.trace("found better plan (" + Format.d(cost, 2) + " < " + Format.d(lowestCost, 2) + "): " + planCount + " of " + equalSet.size() + " in class " + eqClass);
					}
					
					bestPlan = plan;
					lowestCost = cost;
				} else {
					if (LOGGER.isTraceEnabled()) {
						LOGGER.trace("discarding plan (" + Format.d(cost, 2) + " > " + Format.d(lowestCost, 2) + "): " + planCount + " of " + equalSet.size() + " in class " + eqClass);
					}
				}
			}
			
			bestPlans.add(bestPlan);
		}
		
		plans.clear();
		plans.addAll(bestPlans);
		
//		if (LOGGER.isTraceEnabled() && DEBUG_N > 1)
//			LOGGER.trace("reduced to " + bestPlans.size() + " " + DEBUG_N + "-ary join plans.");
	}
	
	class PlanCollection {
		
		private Map<Integer, Set<TupleExpr>> planMap = new HashMap<Integer, Set<TupleExpr>>();
		
		/**
		 * Adds the collection of query plans to the set of n-ary joins.
		 * @param plans the plans to add.
		 * @param n the number of n-ary joins.
		 */
		public void add(Set<TupleExpr> plans, int n) {
			Set<TupleExpr> planSet = planMap.get(n);
			if (planSet == null)
				planMap.put(n, planSet = new HashSet<TupleExpr>());
			planSet.addAll(plans);
		}
		
		/**
		 * Return the collection of query plans for n-ary joins.
		 * @param n the number of n-ary joins.
		 * @return the collection of query plans.
		 */
		public Set<TupleExpr> get(int n) {
			return this.planMap.get(n);
		}
	}

} public class PatternSelectivityOptimizer extends AbstractFederationOptimizer {
	
	@Override
	public TupleExpr optimizeBGP(TupleExpr bgp) {
		
		List<TupleExpr> queryExpressions = this.getBaseExpressions(bgp);
		final Map<TupleExpr, Double> costs = new HashMap<TupleExpr, Double>();
		
		// get cardinality for all query base expressions
		for (TupleExpr expr : queryExpressions) {
			costs.put(expr, this.costEstimator.process(expr));
		}
		
		// defines the cost comparator
		Comparator<TupleExpr> comparator = new Comparator<TupleExpr>() {
			@Override
			public int compare(TupleExpr arg0, TupleExpr arg1) {
				return costs.get(arg0).compareTo(costs.get(arg1));
			}
		};
		
		// create new query by adding expressions with minimum cost first and
		// trying to avoid cross products
		TupleExpr newQuery = null;
		while (!queryExpressions.isEmpty()) {
			
			List<TupleExpr> joinCandidates = new ArrayList<TupleExpr>();
			
			// find join candidates (unless it is a new query)
			if (newQuery != null) {
				// collect all query expressions containing a join variable
				// (to avoid cross products)
				Set<String> queryVars = VarNameCollector.process(newQuery);
				for (TupleExpr expr : queryExpressions) {
					for (String varName : VarNameCollector.process(expr)) {
						if (queryVars.contains(varName)) {
							joinCandidates.add(expr);
							break;
						}
					}
				}
			}
			
			// no candidates if this is a new query or in case of cross product
			if (joinCandidates.size() == 0)
				joinCandidates.addAll(queryExpressions);
			
			// sort expressions by cost
			Collections.sort(joinCandidates, comparator);
			
			// join with selected expression and remove from list
			TupleExpr joinExpr = joinCandidates.get(0);
			newQuery = (newQuery == null) ? joinExpr : new Join(newQuery, joinExpr);
			queryExpressions.remove(joinExpr);
		}
		
		bgp.replaceWith(newQuery);
		return newQuery;
	}

} public class AskSelector extends SourceSelectorBase {
	
	private static final Logger LOGGER = LoggerFactory.getLogger(AskSelector.class);
	
	private List<Graph> sourceList;
	
	@Override
	public void initialize() throws SailException {
		super.initialize();
		this.sourceList = ((VoidStatistics) stats).getEndpoints();
		
		if (LOGGER.isDebugEnabled())
			LOGGER.debug("initialized ask selecector with: " + this.sourceList);
	}

	@Override
	protected Set<Graph> getSources(StatementPattern pattern) {
		return getSources(pattern, this.sourceList);
	}
	
	protected Set<Graph> getSources(StatementPattern pattern, Collection<Graph> sources) {
		Set<Graph> selectedSources = new HashSet<Graph>();
		
		if (LOGGER.isDebugEnabled())
			LOGGER.debug(debugAskRequest(pattern));
		
		String sparqlPattern = OperatorTreePrinter.print(pattern);
		
		// ask each source for current pattern
		for (Graph source : sources) {
			if (QueryExecutor.ask(source.toString(), sparqlPattern))
				selectedSources.add(source);
		}
		return selectedSources;
	}
	
	private String debugAskRequest(StatementPattern pattern) {
		StringBuffer buffer = new StringBuffer("ASK {");
		buffer.append(OperatorTreePrinter.print(pattern));
		buffer.append("} @[");
		for (Graph source : sourceList) {
			buffer.append(source.getNamespaceURL()).append(", ");
		}
		buffer.setLength(buffer.length()-2);
		buffer.append("]");
		return buffer.toString();
	}

} public class IndexAskSelector extends AskSelector {
	
	private IndexSelector indexSel;
	
	/**
	 * Creates a source finder using the supplied statistics and sources.
	 * 
	 * @param stats the statistics to use.
	 */
	public IndexAskSelector(boolean useTypeStats) {
		this.indexSel = new IndexSelector(useTypeStats);
	}
	
	@Override
	public void setStatistics(RDFStatistics stats) {
		super.setStatistics(stats);
		this.indexSel.setStatistics(stats);
	}
	
	@Override
	protected Set<Graph> getSources(StatementPattern pattern) {
		Set<Graph> sources = this.indexSel.getSources(pattern);
		return getSources(pattern, sources);
	}
	
} public class IndexSelector extends SourceSelectorBase {
	
	private boolean useTypeStats = true; 
	
	/**
	 * Creates a source finder using the supplied statistics.
	 * 
	 * @param stats the statistics to use.
	 */
	public IndexSelector(boolean useTypeStats) {
		this.useTypeStats = useTypeStats;
	}

	/**
	 * Find matching sources for the supplied pattern.
	 * 
	 * @param patterns the pattern that need to matched to sources.
	 * @return the set of source which can contribute results for the pattern.
	 */
	@Override
	protected Set<Graph> getSources(StatementPattern pattern) {
		Value s = pattern.getSubjectVar().getValue();
		Value p = pattern.getPredicateVar().getValue();
		Value o = pattern.getObjectVar().getValue();
		
		return stats.findSources(s == null ? null : s.stringValue(),
								 p == null ? null : p.stringValue(),
								 o == null ? null : o.stringValue(), useTypeStats);
	}
	
} public interface SourceSelector {
	
	public void initialize() throws SailException;
	
	public void setStatistics(RDFStatistics stats);
	
	/**
	 * Maps triple patterns to data sources which can contribute results.
	 * 
	 * @param patterns the SPARQL triple patterns which need to be mapped.
	 * @return a list triple patterns with mappings to sources.
	 */
	public List<MappedStatementPattern> mapSources(List<StatementPattern> patterns);

} public class VoidStatistics implements RDFStatistics {
	
	private static final Logger LOGGER = LoggerFactory.getLogger(VoidStatistics.class);
	
	private static final String USER_DIR = System.getProperty("user.dir") + File.separator;
	
	private static final String VOID_PREFIX = "PREFIX void: <" + VOID2.NAMESPACE + ">\n";
	
	private static final String VAR_GRAPH = "$GRAPH$";
	private static final String VAR_TYPE  = "$TYPE$";
	private static final String VAR_PRED  = "$PRED$";

	private static final String PRED_SOURCE = VOID_PREFIX +
			"SELECT ?source WHERE {" +
			"  [] a void:Dataset ;" +
			"     void:sparqlEndpoint ?source ;" +
			"     void:propertyPartition ?part ." +
			"  ?part void:property <" + VAR_PRED + "> ." +
			"}";
	
	private static final String TYPE_SOURCE = VOID_PREFIX +
			"SELECT ?source WHERE {" +
			"  [] a void:Dataset ;" +
			"     void:sparqlEndpoint ?source ;" +
			"     void:classPartition ?part ." +
			"  ?part void:class <" + VAR_TYPE + ">" +
			"}";
	
	private static final String TRIPLE_COUNT = VOID_PREFIX +
			"SELECT ?count WHERE {" +
			"  [] a void:Dataset ;" +
			"     void:triples ?count ;" +
			"     void:sparqlEndpoint <" + VAR_GRAPH + "> ." +
			"}";
	
	private static final String DISTINCT_PREDICATES = VOID_PREFIX +
			"SELECT ?count WHERE {" +
			"  [] a void:Dataset ;" +
			"     void:sparqlEndpoint <" + VAR_GRAPH + "> ;" +
			"     void:properties ?count ." +
			"}";

	private static final String DISTINCT_SUBJECTS = VOID_PREFIX +
			"SELECT ?count WHERE {" +
			"  [] a void:Dataset ;" +
			"     void:sparqlEndpoint <" + VAR_GRAPH + "> ;" +
			"     void:distinctSubjects ?count ." +
			"}";

	private static final String DISTINCT_PRED_SUBJECTS = VOID_PREFIX +
			"SELECT ?count WHERE {" +
			"  [] a void:Dataset ;" +
			"     void:sparqlEndpoint <" + VAR_GRAPH + "> ;" +
			"     void:propertyPartition ?part ." +
			"  ?part void:property <" + VAR_PRED + "> ;" +
			"        void:distinctSubjects ?count ." +
			"}";
	
	private static final String DISTINCT_OBJECTS = VOID_PREFIX +
			"SELECT ?count WHERE {" +
			"  [] a void:Dataset ;" +
			"     void:sparqlEndpoint <" + VAR_GRAPH + "> ;" +
			"     void:distinctObjects ?count ." +
			"}";
	
	private static final String DISTINCT_PRED_OBJECTS = VOID_PREFIX +
			"SELECT ?count WHERE {" +
			"  [] a void:Dataset ;" +
			"     void:sparqlEndpoint <" + VAR_GRAPH + "> ;" +
			"     void:propertyPartition ?part ." +
			"  ?part void:property <" + VAR_PRED + "> ;" +
			"        void:distinctObjects ?count ." +
			"}";
	
	private static final String TYPE_TRIPLES = VOID_PREFIX +
			"SELECT ?count WHERE {" +
			"  [] a void:Dataset ;" +
			"     void:sparqlEndpoint <" + VAR_GRAPH + "> ;" +
			"     void:classPartition ?part ." +
			"  ?part void:class <" + VAR_TYPE + "> ;" +
			"        void:entities ?count ." +
			"}";
	
	private static final String PRED_TRIPLES = VOID_PREFIX +
			"SELECT ?count WHERE {" +
			"  [] a void:Dataset ;" +
			"     void:sparqlEndpoint <" + VAR_GRAPH + "> ;" +
			"     void:propertyPartition ?part ." +
			"  ?part void:property <" + VAR_PRED + "> ;" +
			"        void:triples ?count ." +
			"}";
	
	private static final ValueFactory uf = ValueFactoryImpl.getInstance();
	private static final URI DATASET = uf.createURI(VOID2.Dataset.toString());
	private static final URI ENDPOINT = uf.createURI(VOID2.sparqlEndpoint.toString());
	
	protected static final VoidStatistics singleton = new VoidStatistics();
	
	private final Repository voidRepository;
	
	// --- STATIC -------------------------------------------------------------

	public static VoidStatistics getInstance() {
		return singleton;
	}
	
	// --- PRIVATE ------------------------------------------------------------
	
	/**
	 * Private constructor used for singleton creation.
	 */
	private VoidStatistics() {
		this.voidRepository = new SailRepository(new MemoryStore());
		try {
			this.voidRepository.initialize();
		} catch (RepositoryException e) {
			throw new RuntimeException("initialization of statistics repository failed", e);
		}
	}
	
	/**
	 * Returns the count value defined by the supplied query and variable substitutions.
	 * 
	 * @param query the query to be executed on the voiD repository.
	 * @param vars the variable bindings to be substituted in the query.
	 * @return the resulting count value.
	 */
	private long getCount(String query, String... vars) {
		
		// replace query variables
		for (int i = 0; i < vars.length; i++) {
			query = query.replace(vars[i], vars[++i]);
		}
		
		List<String> bindings = evalQuery(query, "count");
		
		// check result validity
		if (bindings.size() == 0) {
			LOGGER.warn("found no count for " + Arrays.asList(vars));
			return -1;
		}
		if (bindings.size() > 1)
			LOGGER.warn("found multiple counts for " + Arrays.asList(vars));
		
		return Long.parseLong(bindings.get(0));
	}
	
	/**
	 * Evaluates the given query and returns the result values for the specified binding name.
	 * 
	 * @param query the query to evaluate.
	 * @param bindingName the desired result bindings.
	 * @return a list of result binding values.
	 */
	private List<String> evalQuery(String query, String bindingName) {
		try {
			RepositoryConnection con = this.voidRepository.getConnection();
			try {
				TupleQuery tupleQuery = con.prepareTupleQuery(SPARQL, query);
				TupleQueryResult result = tupleQuery.evaluate();
				
				try {
					List<String> bindings = new ArrayList<String>();
					while (result.hasNext()) {
						bindings.add(result.next().getValue(bindingName).stringValue());
					}
					return bindings;
				} catch (QueryEvaluationException e) {
					LOGGER.error("failed to handle query result from voiD repository, " + e.getMessage() + "\n" + query, e);
				} finally {
					result.close();
				}
			} catch (IllegalArgumentException e) {
				LOGGER.error("not a tuple query, " + e.getMessage() + "\n" + query, e);	
			} catch (RepositoryException e) {
				LOGGER.error("failed to create tuple query, " + e.getMessage() + "\n" + query, e);
			} catch (MalformedQueryException e) {
				LOGGER.error("malformed query, " + e.getMessage() + "\n" + query, e);
			} catch (QueryEvaluationException e) {
				LOGGER.error("failed to evaluate query on voiD repository, " + e.getMessage() + "\n" + query, e);
			} finally {
				con.close();
			}
		} catch (RepositoryException e) {
			LOGGER.error("failed to open/close voiD repository connection, " + e.getMessage() + "\n" + query, e);
		}
		return null;
	}
	
	private List<URI> getEndpoints(URI voidURI, RepositoryConnection con) throws RepositoryException {
		ValueFactory uf = this.voidRepository.getValueFactory();
		URI voidDataset = uf.createURI(VOID2.Dataset.toString());
		URI sparqlEndpoint = uf.createURI(VOID2.sparqlEndpoint.toString());
		List<URI> endpoints = new ArrayList<URI>();
		
		for (Statement dataset : con.getStatements(null, RDF.TYPE, voidDataset, false, voidURI).asList()) {
			for (Statement endpoint : con.getStatements(dataset.getSubject(), sparqlEndpoint, null, false, voidURI).asList()) {
				// TODO: endpoint may be a literal
				endpoints.add((URI) endpoint.getObject());
			}
		}
		return endpoints;
	}
	
	// -------------------------------------------------------------------------
	
	@Override
	public Set<Graph> findSources(String sValue, String pValue, String oValue, boolean handleType) {
		
		Set<Graph> sources = new HashSet<Graph>();
		
		if (pValue == null) {
			LOGGER.info("found triple pattern with unbound predicate: selecting all sources");
			sources.addAll(getEndpoints());
			return sources;
		}
		
		String query = null;
		// query for RDF type occurrence if rdf:type with bound object is used
		if (handleType && RDF.TYPE.stringValue().equals(pValue) && oValue != null) {
			query = TYPE_SOURCE.replace(VAR_TYPE, oValue);
		} else { // else query for predicate occurrence
			query = PRED_SOURCE.replace(VAR_PRED, pValue);
		}
		
		// execute query and get all source bindings
		for (String graph : evalQuery(query, "source")) {
			sources.add(new Graph(graph));
		}
		return sources;
	}
	
	@Override
	public long getTripleCount(Graph g) {
		return getCount(TRIPLE_COUNT, VAR_GRAPH, g.toString());
	}
	
	@Override
	public long getPredicateCount(Graph g, String predicate) {
		return getCount(PRED_TRIPLES, VAR_GRAPH, g.toString(), VAR_PRED, predicate);
	}
	
	@Override
	public long getTypeCount(Graph g, String type) {
		return getCount(TYPE_TRIPLES, VAR_GRAPH, g.toString(), VAR_TYPE, type);
	}
	
	@Override
	public long getDistinctPredicates(Graph g) {
		return getCount(DISTINCT_PREDICATES, VAR_GRAPH, g.toString());
	}
	
	@Override
	public long getDistinctSubjects(Graph g) {
		return getCount(DISTINCT_SUBJECTS, VAR_GRAPH, g.toString());
	}
	
	@Override
	public long getDistinctSubjects(Graph g, String predicate) {
		return getCount(DISTINCT_PRED_SUBJECTS, VAR_GRAPH, g.toString(), VAR_PRED, predicate);
	}
	
	@Override
	public long getDistinctObjects(Graph g) {
		return getCount(DISTINCT_OBJECTS, VAR_GRAPH, g.toString());
	}

	@Override
	public long getDistinctObjects(Graph g, String predicate) {
		return getCount(DISTINCT_PRED_OBJECTS, VAR_GRAPH, g.toString(), VAR_PRED, predicate);
	}
	
	// -------------------------------------------------------------------------
	
	/**
	 * Extracts all SPARQL endpoints of the datasets.
	 * 
	 * @return the list of SPARQL endpoints.
	 */
	public List<Graph> getEndpoints() {
		
		List<Graph> sources = new ArrayList<Graph>();
		
		ValueFactory uf = this.voidRepository.getValueFactory();
		URI voidDataset = uf.createURI(VOID2.Dataset.toString());
		URI sparqlEndpoint = uf.createURI(VOID2.sparqlEndpoint.toString());
		
		try {
			RepositoryConnection con = this.voidRepository.getConnection();
			
			try {
				for (Statement dataset : con.getStatements(null, RDF.TYPE, voidDataset, false).asList()) {
					for (Statement endpoint : con.getStatements(dataset.getSubject(), sparqlEndpoint, null, false).asList()) {
						sources.add(new Graph(endpoint.getObject().stringValue()));
					}
				}
			} catch (RepositoryException e) {
				e.printStackTrace();
			} finally {
				con.close();
			}
		} catch (RepositoryException e) {
			e.printStackTrace();
		}
		
		return sources;
	}
	
	/**
	 * Loads the supplied voiD description into the statistics repository.
	 * 
	 * @param voidURI the URI of the voiD description to load.
	 * @return the assigned SPARQL endpoint.
	 */
	public URI load(URI voidURI, URI endpoint) throws IOException {
		if (voidURI == null)
			throw new IllegalArgumentException("voiD URI must not be null.");
		
		// initialize parser
		RDFFormat format = Rio.getParserFormatForFileName(voidURI.stringValue());
		if (format == null) {
			throw new IOException("Unsupported RDF format: " + voidURI);
		}

		URL voidURL = new URL(voidURI.stringValue());  // throws IOException
		InputStream in = voidURL.openStream();
		try {
			
			RepositoryConnection con = this.voidRepository.getConnection();
			try {
				
				// check if voiD description has already been loaded
				List<URI> endpoints = getEndpoints(voidURI, con);
				if (endpoints.size() > 0) {
					LOGGER.warn("VOID has already been loaded: " + voidURI);
					return endpoints.get(0);
				}
				
				// add voiD file content to repository
				try {
					con.add(in, voidURI.stringValue(), format, voidURI);
				} catch (RDFParseException e) {
					LOGGER.error("can not parse VOID file " + voidURI + ": " + e.getMessage());
					return null;
				} catch (RepositoryException e) {
					LOGGER.error("can not add VOID file: " + voidURI, e);
					return null;
				}
				
				if (LOGGER.isDebugEnabled())
					LOGGER.debug("loaded VOID: " + voidURL.getPath().replace(USER_DIR, ""));
				
				if (endpoint == null) {
				
					// check if this voiD description has a valid SPARQL endpoint
					endpoints = getEndpoints(voidURI, con);

					if (endpoints.size() == 0)
						LOGGER.debug("found no SPARQL endpoint in voiD file");
					if (endpoints.size() > 1)
						// TODO: don't throw Exception but use first endpoint only
						throw new IllegalStateException("found multiple SPARQL endpoints in voiD file");

				return endpoints.iterator().next();
				} else {
					// find dataset resource in specified context
					RepositoryResult<Statement> result = con.getStatements(null, RDF.TYPE, DATASET, false, voidURI);
					Resource dataset = result.next().getSubject();
					
					// TODO: check that there is only one dataset defined
					
					// remove current SPARQL endpoint and add new one
					con.remove(dataset, ENDPOINT, null, voidURI);
					con.add(dataset, ENDPOINT, endpoint, voidURI);
					
					LOGGER.info("set SPARQL endpoint '" + endpoint + "' for " + voidURL.getPath().replace(USER_DIR, ""));
					
					return endpoint;
				}
				
			} catch (RepositoryException e) {
				e.printStackTrace();
//			} catch (RDFParseException e) {
//				e.printStackTrace();
//			} catch (IOException e) {
//				e.printStackTrace();
			} finally {
				con.close();
			}
		} catch (RepositoryException e) {
			e.printStackTrace();
		} finally {
			in.close();
		}
		
		return null;
	}
	
}
import org.openrdf.repository.config.RepositoryConfigException;
import org.openrdf.repository.config.RepositoryFactory;
import org.openrdf.repository.config.RepositoryImplConfig;
import org.openrdf.repository.config.RepositoryRegistry;
import org.openrdf.sail.Sail;
import org.openrdf.sail.config.SailConfigException;
import org.openrdf.sail.config.SailFactory;
import org.openrdf.sail.config.SailImplConfig;
import de.uni_koblenz.west.splendid.FederationSail;
import de.uni_koblenz.west.splendid.estimation.AbstractCardinalityEstimator;
import de.uni_koblenz.west.splendid.estimation.AbstractCostEstimator;
import de.uni_koblenz.west.splendid.estimation.CardinalityCostEstimator;
import de.uni_koblenz.west.splendid.estimation.ModelEvaluator;
import de.uni_koblenz.west.splendid.estimation.SPLENDIDCardinalityEstimator;
import de.uni_koblenz.west.splendid.estimation.SPLENDIDCostEstimator;
import de.uni_koblenz.west.splendid.estimation.TrueCardinalityEstimator;
import de.uni_koblenz.west.splendid.estimation.VoidCardinalityEstimator;
import de.uni_koblenz.west.splendid.model.SubQueryBuilder;
import de.uni_koblenz.west.splendid.optimizer.AbstractFederationOptimizer;
import de.uni_koblenz.west.splendid.optimizer.DynamicProgrammingOptimizer;
import de.uni_koblenz.west.splendid.optimizer.PatternSelectivityOptimizer;
import de.uni_koblenz.west.splendid.sources.AskSelector;
import de.uni_koblenz.west.splendid.sources.IndexAskSelector;
import de.uni_koblenz.west.splendid.sources.IndexSelector;
import de.uni_koblenz.west.splendid.sources.SourceSelector;
import de.uni_koblenz.west.splendid.statistics.VoidStatistics;
/*
 * This file is part of RDF Federator.
 * Copyright 2010 Olaf Goerlitz
 * 
 * RDF Federator is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 * 
 * RDF Federator is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Lesser General Public License for more details.
 * 
 * You should have received a copy of the GNU Lesser General Public License
 * along with RDF Federator.  If not, see <http://www.gnu.org/licenses/>.
 * 
 * RDF Federator uses libraries from the OpenRDF Sesame Project licensed 
 * under the Aduna BSD-style license. 
 */
package de.uni_koblenz.west.splendid.config;



/**
 * A {@link SailFactory} that creates {@link FederationSail}s
 * based on the supplied configuration data.
 * 
 * ATTENTION: This factory must be published with full package name in
 *            META-INF/services/org.openrdf.sail.config.SailFactory
 * 
 * @author Olaf Goerlitz
 */
public class FederationSailFactory implements SailFactory {
	
	/**
	 * The type of repositories that are created by this factory.
	 * 
	 * @see SailFactory#getSailType()
	 */
	public static final String SAIL_TYPE = "west:FederationSail";

	/**
	 * Returns the Sail's type: <tt>west:FederationSail</tt>.
	 * 
	 * @return the Sail's type.
	 */
	public String getSailType() {
		return SAIL_TYPE;
	}

	/**
	 * Provides a Sail configuration object for the configuration data.
	 * 
	 * @return a {@link FederationSailConfig}.
	 */
	public SailImplConfig getConfig() {
		return new FederationSailConfig();
	}

	/**
	 * Returns a Sail instance that has been initialized using the supplied
	 * configuration data.
	 * 
	 * @param config
	 *            the Sail configuration.
	 * @return The created (but un-initialized) Sail.
	 * @throws SailConfigException
	 *             If no Sail could be created due to invalid or incomplete
	 *             configuration data.
	 */
	@Override
	public Sail getSail(SailImplConfig config) throws SailConfigException { // Sesame 2

		if (!SAIL_TYPE.equals(config.getType())) {
			throw new SailConfigException("Invalid Sail type: " + config.getType());
		}
		
		RepositoryRegistry registry = RepositoryRegistry.getInstance();
		
		assert config instanceof FederationSailConfig;
		FederationSailConfig cfg = (FederationSailConfig)config;
		FederationSail sail = new FederationSail();
		
		// Create all member repositories
		for (RepositoryImplConfig repConfig : cfg.getMemberConfigs()) {
			RepositoryFactory factory = registry.get(repConfig.getType());
			if (factory == null) {
				throw new SailConfigException("Unsupported repository type: " + repConfig.getType());
			}
			try {
				sail.addMember(factory.getRepository(repConfig));
			} catch (RepositoryConfigException e) {
				throw new SailConfigException("invalid repository configuration: " + e.getMessage(), e);
			}
		}

		// create query optimizer
		QueryOptimizerConfig optConfig = cfg.getOptimizerConfig(); 
		AbstractFederationOptimizer opt = getQueryOptimizer(optConfig);
		sail.setFederationOptimizer(opt);
		
		// create evaluation strategy
		sail.setEvalStrategy(optConfig.getEvalStrategy());
		
		// setup statistics
		boolean voidPlus = true;
		String estType = optConfig.getEstimatorType();
		if ("VOID".equalsIgnoreCase(estType))
			voidPlus = false;
		if ("VOID_PLUS".equalsIgnoreCase(estType))
			voidPlus = true;
		
		VoidStatistics stats = VoidStatistics.getInstance();
		AbstractCardinalityEstimator cardEstim = new SPLENDIDCardinalityEstimator(stats, voidPlus);
		AbstractCostEstimator costEstim = new SPLENDIDCostEstimator();
		costEstim.setCardinalityEstimator(cardEstim);
//		ModelEvaluator modelEval = new TrueCardinalityEstimator(sail.getEvalStrategy());
		
		// Create source selector from configuration settings
		SourceSelector selector = getSourceSelector(cfg.getSelectorConfig());
		sail.setSourceSelector(selector);
		
		opt.setBuilder(new SubQueryBuilder(optConfig));
		opt.setSelector(selector);
		opt.setCostEstimator(costEstim);
//		opt.setModelEvaluator(cardEstim);
		opt.setModelEvaluator(costEstim);
//		opt.setModelEvaluator(modelEval);
		

		return sail;
	}
	
	// --------------------------------------------------------------
	
	/**
	 * Creates a sources selector for the given configuration settings.
	 * 
	 * @param selConf the source selector configuration settings.
	 * @return the created sources selector.
	 * @throws SailConfigException If no source selector could be created due to invalid or incomplete
	 *             configuration data.
	 */
	private SourceSelector getSourceSelector(SourceSelectorConfig selConf) throws SailConfigException {
		String selectorType = selConf.getType();
		
		if ("ASK".equalsIgnoreCase(selectorType))
			return new AskSelector();
		else if ("INDEX".equalsIgnoreCase(selectorType))
			return new IndexSelector(selConf.isUseTypeStats());
		else if ("INDEX_ASK".equalsIgnoreCase(selectorType))
			return new IndexAskSelector(selConf.isUseTypeStats());
		
		throw new SailConfigException("invalid source selector type: " + selectorType);
	}
	
	/**
	 * Creates a query optimizer for the given configuration settings.
	 *  
	 * @param optConf the query optimizer configuration settings.
	 * @return the created query optimizer.
	 * @throws SailConfigException If no query optimizer could be created due to invalid or incomplete
	 *             configuration data.
	 */
	private AbstractFederationOptimizer getQueryOptimizer(QueryOptimizerConfig optConf) throws SailConfigException {
		String optimizerType = optConf.getType();
		
		if ("DYNAMIC_PROGRAMMING".equals(optimizerType))
			return new DynamicProgrammingOptimizer(optConf.isUseHashJoin(), optConf.isUseBindJoin());